diff --git a/logstash-core/lib/logstash/environment.rb b/logstash-core/lib/logstash/environment.rb
index 3c2bd7377..e3a04b27c 100644
--- a/logstash-core/lib/logstash/environment.rb
+++ b/logstash-core/lib/logstash/environment.rb
@@ -20,7 +20,6 @@ require "logstash/config/cpu_core_strategy"
 require "logstash/settings"
 require "logstash/util/cloud_setting_id"
 require "logstash/util/cloud_setting_auth"
-require "logstash/util/modules_setting_array"
 require "socket"
 require "stud/temporary"

@@ -34,26 +33,18 @@ module LogStash
   end

   [
-           Setting::Boolean.new("allow_superuser", true),
-            Setting::String.new("node.name", Socket.gethostname),
-    Setting::NullableString.new("path.config", nil, false),
+           Setting::Boolean.new("allow_superuser", false),
+            Setting::SettingString.new("node.name", Socket.gethostname),
+    Setting::SettingNullableString.new("path.config", nil, false),
  Setting::WritableDirectory.new("path.data", ::File.join(LogStash::Environment::LOGSTASH_HOME, "data")),
-    Setting::NullableString.new("config.string", nil, false),
-           Setting::Modules.new("modules.cli", LogStash::Util::ModulesSettingArray, []),
-           Setting::Modules.new("modules", LogStash::Util::ModulesSettingArray, []),
-                    Setting.new("modules_list", Array, []),
-                    Setting.new("modules_variable_list", Array, []),
-           Setting::Modules.new("cloud.id", LogStash::Util::CloudSettingId),
-           Setting::Modules.new("cloud.auth", LogStash::Util::CloudSettingAuth),
-           Setting::Boolean.new("modules_setup", false),
+    Setting::SettingNullableString.new("config.string", nil, false),
            Setting::Boolean.new("config.test_and_exit", false),
            Setting::Boolean.new("config.reload.automatic", false),
          Setting::TimeValue.new("config.reload.interval", "3s"), # in seconds
            Setting::Boolean.new("config.support_escapes", false),
-            Setting::String.new("config.field_reference.escape_style", "none", true, %w(none percent ampersand)),
-            Setting::String.new("event_api.tags.illegal", "rename", true, %w(rename warn)),
+            Setting::SettingString.new("config.field_reference.escape_style", "none", true, %w(none percent ampersand)),
            Setting::Boolean.new("metric.collect", true),
-            Setting::String.new("pipeline.id", "main"),
+            Setting::SettingString.new("pipeline.id", "main"),
            Setting::Boolean.new("pipeline.system", false),
    Setting::PositiveInteger.new("pipeline.workers", LogStash::Config::CpuCoreStrategy.maximum),
    Setting::PositiveInteger.new("pipeline.batch.size", 125),
@@ -65,32 +56,32 @@ module LogStash
    Setting::CoercibleString.new("pipeline.ordered", "auto", true, ["auto", "true", "false"]),
    Setting::CoercibleString.new("pipeline.ecs_compatibility", "v8", true, %w(disabled v1 v8)),
                     Setting.new("path.plugins", Array, []),
-    Setting::NullableString.new("interactive", nil, false),
+    Setting::SettingNullableString.new("interactive", nil, false),
            Setting::Boolean.new("config.debug", false),
-            Setting::String.new("log.level", "info", true, ["fatal", "error", "warn", "debug", "info", "trace"]),
+            Setting::SettingString.new("log.level", "info", true, ["fatal", "error", "warn", "debug", "info", "trace"]),
            Setting::Boolean.new("version", false),
            Setting::Boolean.new("help", false),
            Setting::Boolean.new("enable-local-plugin-development", false),
-            Setting::String.new("log.format", "plain", true, ["json", "plain"]),
-           Setting::Boolean.new("log.format.json.fix_duplicate_message_fields", false),
-           Setting::Boolean.new("api.enabled", true).with_deprecated_alias("http.enabled"),
-            Setting::String.new("api.http.host", "127.0.0.1").with_deprecated_alias("http.host"),
-         Setting::PortRange.new("api.http.port", 9600..9700).with_deprecated_alias("http.port"),
-            Setting::String.new("api.environment", "production").with_deprecated_alias("http.environment"),
-            Setting::String.new("api.auth.type", "none", true, %w(none basic)),
-            Setting::String.new("api.auth.basic.username", nil, false).nullable,
+            Setting::SettingString.new("log.format", "plain", true, ["json", "plain"]),
+           Setting::Boolean.new("log.format.json.fix_duplicate_message_fields", true),
+           Setting::Boolean.new("api.enabled", true),
+            Setting::SettingString.new("api.http.host", "127.0.0.1"),
+         Setting::PortRange.new("api.http.port", 9600..9700),
+            Setting::SettingString.new("api.environment", "production"),
+            Setting::SettingString.new("api.auth.type", "none", true, %w(none basic)),
+            Setting::SettingString.new("api.auth.basic.username", nil, false).nullable,
           Setting::Password.new("api.auth.basic.password", nil, false).nullable,
-            Setting::String.new("api.auth.basic.password_policy.mode", "WARN", true, %w[WARN ERROR]),
+            Setting::SettingString.new("api.auth.basic.password_policy.mode", "WARN", true, %w[WARN ERROR]),
            Setting::Numeric.new("api.auth.basic.password_policy.length.minimum", 8),
-            Setting::String.new("api.auth.basic.password_policy.include.upper", "REQUIRED", true, %w[REQUIRED OPTIONAL]),
-            Setting::String.new("api.auth.basic.password_policy.include.lower", "REQUIRED", true, %w[REQUIRED OPTIONAL]),
-            Setting::String.new("api.auth.basic.password_policy.include.digit", "REQUIRED", true, %w[REQUIRED OPTIONAL]),
-            Setting::String.new("api.auth.basic.password_policy.include.symbol", "OPTIONAL", true, %w[REQUIRED OPTIONAL]),
+            Setting::SettingString.new("api.auth.basic.password_policy.include.upper", "REQUIRED", true, %w[REQUIRED OPTIONAL]),
+            Setting::SettingString.new("api.auth.basic.password_policy.include.lower", "REQUIRED", true, %w[REQUIRED OPTIONAL]),
+            Setting::SettingString.new("api.auth.basic.password_policy.include.digit", "REQUIRED", true, %w[REQUIRED OPTIONAL]),
+            Setting::SettingString.new("api.auth.basic.password_policy.include.symbol", "OPTIONAL", true, %w[REQUIRED OPTIONAL]),
            Setting::Boolean.new("api.ssl.enabled", false),
   Setting::ExistingFilePath.new("api.ssl.keystore.path", nil, false).nullable,
           Setting::Password.new("api.ssl.keystore.password", nil, false).nullable,
        Setting::StringArray.new("api.ssl.supported_protocols", nil, true, %w[TLSv1 TLSv1.1 TLSv1.2 TLSv1.3]),
-            Setting::String.new("queue.type", "memory", true, ["persisted", "memory"]),
+            Setting::SettingString.new("queue.type", "memory", true, ["persisted", "memory"]),
            Setting::Boolean.new("queue.drain", false),
              Setting::Bytes.new("queue.page_capacity", "64mb"),
              Setting::Bytes.new("queue.max_bytes", "1024mb"),
@@ -102,16 +93,16 @@ module LogStash
            Setting::Boolean.new("dead_letter_queue.enable", false),
              Setting::Bytes.new("dead_letter_queue.max_bytes", "1024mb"),
            Setting::Numeric.new("dead_letter_queue.flush_interval", 5000),
-            Setting::String.new("dead_letter_queue.storage_policy", "drop_newer", true, ["drop_newer", "drop_older"]),
-    Setting::NullableString.new("dead_letter_queue.retain.age"), # example 5d
+            Setting::SettingString.new("dead_letter_queue.storage_policy", "drop_newer", true, ["drop_newer", "drop_older"]),
+    Setting::SettingNullableString.new("dead_letter_queue.retain.age"), # example 5d
          Setting::TimeValue.new("slowlog.threshold.warn", "-1"),
          Setting::TimeValue.new("slowlog.threshold.info", "-1"),
          Setting::TimeValue.new("slowlog.threshold.debug", "-1"),
          Setting::TimeValue.new("slowlog.threshold.trace", "-1"),
-            Setting::String.new("keystore.classname", "org.logstash.secret.store.backend.JavaKeyStore"),
-            Setting::String.new("keystore.file", ::File.join(::File.join(LogStash::Environment::LOGSTASH_HOME, "config"), "logstash.keystore"), false), # will be populated on
-    Setting::NullableString.new("monitoring.cluster_uuid"),
-            Setting::String.new("pipeline.buffer.type", nil, false, ["direct", "heap"])
+            Setting::SettingString.new("keystore.classname", "org.logstash.secret.store.backend.JavaKeyStore"),
+            Setting::SettingString.new("keystore.file", ::File.join(::File.join(LogStash::Environment::LOGSTASH_HOME, "config"), "logstash.keystore"), false), # will be populated on
+    Setting::SettingNullableString.new("monitoring.cluster_uuid"),
+            Setting::SettingString.new("pipeline.buffer.type", "heap", true, ["direct", "heap"])
   # post_process
   ].each {|setting| SETTINGS.register(setting) }


diff --git a/logstash-core/lib/logstash/settings.rb b/logstash-core/lib/logstash/settings.rb
index 5d2fbd588..3696886cd 100644
--- a/logstash-core/lib/logstash/settings.rb
+++ b/logstash-core/lib/logstash/settings.rb
@@ -334,8 +334,8 @@ module LogStash
       validate(value)
     end

-    def with_deprecated_alias(deprecated_alias_name)
-      SettingWithDeprecatedAlias.wrap(self, deprecated_alias_name)
+    def with_deprecated_alias(deprecated_alias_name, obsoleted_version=nil)
+      SettingWithDeprecatedAlias.wrap(self, deprecated_alias_name, obsoleted_version)
     end

     ##
@@ -523,27 +523,10 @@ module LogStash
         @validator_class.validate(value)
       end
     end
+
+    java_import org.logstash.settings.SettingString

-    class String < Setting
-      def initialize(name, default = nil, strict = true, possible_strings = [])
-        @possible_strings = possible_strings
-        super(name, ::String, default, strict)
-      end
-
-      def validate(value)
-        super(value)
-        unless @possible_strings.empty? || @possible_strings.include?(value)
-          raise ArgumentError.new("Invalid value \"#{name}: #{value}\". Options are: #{@possible_strings.inspect}")
-        end
-      end
-    end
-
-    class NullableString < String
-      def validate(value)
-        return if value.nil?
-        super(value)
-      end
-    end
+    java_import org.logstash.settings.SettingNullableString

     class Password < Coercible
       def initialize(name, default = nil, strict = true)
@@ -800,29 +783,7 @@ module LogStash
       end
     end

-    class Modules < Coercible
-      def initialize(name, klass, default = nil)
-        super(name, klass, default, false)
-      end
-
-      def set(value)
-        coerced_value = coerce(value)
-        @wrapped_setting.set(coerced_value)
-        coerced_value
-      end
-
-      def coerce(value)
-        if value.is_a?(@klass)
-          return value
-        end
-        @klass.new(value)
-      end
-
-      protected
-      def validate(value)
-        coerce(value)
-      end
-    end
+    java_import org.logstash.settings.NullableSetting

     # @see Setting#nullable
     # @api internal
@@ -847,10 +808,11 @@ module LogStash
     class DeprecatedAlias < SimpleDelegator
       # include LogStash::Util::Loggable
       alias_method :wrapped, :__getobj__
-      attr_reader :canonical_proxy
+      attr_reader :canonical_proxy, :obsoleted_version

-      def initialize(canonical_proxy, alias_name)
+      def initialize(canonical_proxy, alias_name, obsoleted_version)
         @canonical_proxy = canonical_proxy
+        @obsoleted_version = obsoleted_version

         clone = @canonical_proxy.canonical_setting.clone
         clone.update_wrapper(clone.wrapped_setting.deprecate(alias_name))
@@ -882,9 +844,15 @@ module LogStash
       private

       def do_log_setter_deprecation
-        deprecation_logger.deprecated(I18n.t("logstash.settings.deprecation.set",
-                                             :deprecated_alias => name,
-                                             :canonical_name => canonical_proxy.name))
+        deprecation_logger.deprecated(
+          I18n.t("logstash.settings.deprecation.set",
+                 :deprecated_alias => name,
+                 :canonical_name => canonical_proxy.name,
+                 :obsoleted_sentences =>
+                   @obsoleted_version.nil? ?
+                     I18n.t("logstash.settings.deprecation.obsoleted_future") :
+                     I18n.t("logstash.settings.deprecation.obsoleted_version", :obsoleted_version => @obsoleted_version))
+        )
       end
     end

@@ -901,10 +869,11 @@ module LogStash
       # including the canonical setting and a deprecated alias.
       # @param canonical_setting [Setting]: the setting to wrap
       # @param deprecated_alias_name [String]: the name for the deprecated alias
+      # @param obsoleted_version [String]: the version of Logstash that deprecated alias will be removed
       #
       # @return [SettingWithDeprecatedAlias,DeprecatedSetting]
-      def self.wrap(canonical_setting, deprecated_alias_name)
-        setting_proxy = new(canonical_setting, deprecated_alias_name)
+      def self.wrap(canonical_setting, deprecated_alias_name, obsoleted_version=nil)
+        setting_proxy = new(canonical_setting, deprecated_alias_name, obsoleted_version)

         [setting_proxy, setting_proxy.deprecated_alias]
       end
@@ -912,10 +881,10 @@ module LogStash
       attr_reader :deprecated_alias
       alias_method :canonical_setting, :__getobj__

-      def initialize(canonical_setting, deprecated_alias_name)
+      def initialize(canonical_setting, deprecated_alias_name, obsoleted_version)
         super(canonical_setting)

-        @deprecated_alias = DeprecatedAlias.new(self, deprecated_alias_name)
+        @deprecated_alias = DeprecatedAlias.new(self, deprecated_alias_name, obsoleted_version)
       end

       def set(value)




diff --git a/logstash-core/spec/logstash/settings/setting_with_deprecated_alias_spec.rb b/logstash-core/spec/logstash/settings/setting_with_deprecated_alias_spec.rb
index 481e387ba..4ba04fdff 100644
--- a/logstash-core/spec/logstash/settings/setting_with_deprecated_alias_spec.rb
+++ b/logstash-core/spec/logstash/settings/setting_with_deprecated_alias_spec.rb
@@ -25,15 +25,44 @@ describe LogStash::Setting::SettingWithDeprecatedAlias do
   let(:default_value) { "DeFaUlT" }

   let(:settings) { LogStash::Settings.new }
-  let(:canonical_setting) { LogStash::Setting::String.new(canonical_setting_name, default_value, true) }
+  let(:canonical_setting) { LogStash::Setting::SettingString.new(canonical_setting_name, default_value, true) }
+
+  let(:events) { [] }

   before(:each) do
+    java_import org.apache.logging.log4j.LogManager
+    logger = LogManager.getLogger("org.logstash.settings.DeprecatedAlias")
+    deprecated_logger = LogManager.getLogger("org.logstash.deprecation.settings.DeprecatedAlias")
+
+    @custom_appender = CustomAppender.new(events).tap {|appender| appender.start }
+
+    java_import org.apache.logging.log4j.Level
+    logger.addAppender(@custom_appender)
+    deprecated_logger.addAppender(@custom_appender)
+    # had to set level after appending as it was "error" for some reason
+    logger.setLevel(Level::INFO)
+    deprecated_logger.setLevel(Level::INFO)
+
+    expect(@custom_appender.started?).to be_truthy
+
     allow(LogStash::Settings).to receive(:logger).and_return(double("SettingsLogger").as_null_object)
     allow(LogStash::Settings).to receive(:deprecation_logger).and_return(double("SettingsDeprecationLogger").as_null_object)

     settings.register(canonical_setting.with_deprecated_alias(deprecated_setting_name))
   end

+  after(:each) do
+    events.clear
+    java_import org.apache.logging.log4j.LogManager
+    logger = LogManager.getLogger("org.logstash.settings.DeprecatedAlias")
+    deprecated_logger = LogManager.getLogger("org.logstash.deprecation.settings.DeprecatedAlias")
+    # The Logger's AbstractConfiguration contains a cache of Appender, by class name. The cache is updated
+    # iff is absent, so to make subsequent add_appender effective we have cleanup on teardown, else the new
+    # appender instance is simply not used by the logger
+    logger.remove_appender(@custom_appender)
+    deprecated_logger.remove_appender(@custom_appender)
+  end
+
   shared_examples '#validate_value success' do
     context '#validate_value' do
       it "returns without raising" do
@@ -57,6 +86,7 @@ describe LogStash::Setting::SettingWithDeprecatedAlias do
       it 'does not emit a deprecation warning' do
         expect(LogStash::Settings.deprecation_logger).to_not receive(:deprecated).with(a_string_including(deprecated_setting_name))
         settings.get_setting(deprecated_setting_name).observe_post_process
+        expect(events).to be_empty
       end
     end
   end
@@ -66,6 +96,7 @@ describe LogStash::Setting::SettingWithDeprecatedAlias do

     before(:each) do
       settings.set(deprecated_setting_name, value)
+      settings.get_setting(deprecated_setting_name).observe_post_process
     end

     it 'resolves to the value provided for the deprecated alias' do
@@ -73,15 +104,15 @@ describe LogStash::Setting::SettingWithDeprecatedAlias do
     end

     it 'logs a deprecation warning' do
-      expect(LogStash::Settings.deprecation_logger).to have_received(:deprecated).with(a_string_including(deprecated_setting_name))
+      expect(events[0]).to include(deprecated_setting_name)
     end

     include_examples '#validate_value success'

     context "#observe_post_process" do
       it 're-emits the deprecation warning' do
-        expect(LogStash::Settings.deprecation_logger).to receive(:deprecated).with(a_string_including(deprecated_setting_name))
         settings.get_setting(deprecated_setting_name).observe_post_process
+        expect(events[0]).to include(deprecated_setting_name)
       end
     end

@@ -105,6 +136,39 @@ describe LogStash::Setting::SettingWithDeprecatedAlias do
         expect { settings.get_setting(canonical_setting_name).deprecated_alias.validate_value }.to_not raise_error
       end
     end
+
+    context 'obsoleted version' do
+      before(:each) do
+        settings.register(subject.with_deprecated_alias(deprecated_name, "9"))
+      end
+
+      describe "ruby string setting" do
+        let(:new_value) { "ironman" }
+        let(:old_value) { "iron man" }
+        let(:canonical_name) { "iron.setting" }
+        let(:deprecated_name) { "iron.oxide.setting" }
+        subject { LogStash::Setting::SettingString.new(canonical_name, old_value, true) }
+
+        it 'logs a deprecation warning with target remove version' do
+          settings.set(deprecated_name, new_value)
+          settings.get_setting(deprecated_name).observe_post_process
+          expect(events.length).to be 2
+          expect(events[1]).to include(deprecated_name)
+          expect(events[1]).to include("version 9")
+        end
+      end
+      describe "java boolean setting" do
+        let(:new_value) { false }
+        let(:old_value) { true }
+        let(:canonical_name) { "bool.setting" }
+        let(:deprecated_name) { "boo.setting" }
+        subject { LogStash::Setting::Boolean.new(canonical_name, old_value, true) }
+
+        it 'does not raise error' do
+          expect { settings.set(deprecated_name, new_value) }.to_not raise_error
+        end
+      end
+    end
   end

   context "when only the canonical setting is set" do
@@ -117,15 +181,16 @@ describe LogStash::Setting::SettingWithDeprecatedAlias do
     end

     it 'does not produce a relevant deprecation warning' do
-      expect(LogStash::Settings.deprecation_logger).to_not have_received(:deprecated).with(a_string_including(deprecated_setting_name))
+      settings.get_setting(deprecated_setting_name).observe_post_process
+      expect(events).to be_empty
     end

     include_examples '#validate_value success'

     context "#observe_post_process" do
       it 'does not emit a deprecation warning' do
-        expect(LogStash::Settings.deprecation_logger).to_not receive(:deprecated).with(a_string_including(deprecated_setting_name))
         settings.get_setting(deprecated_setting_name).observe_post_process
+        expect(events).to be_empty
       end
     end
   end
@@ -139,15 +204,15 @@ describe LogStash::Setting::SettingWithDeprecatedAlias do
     context '#validate_value' do
       it "raises helpful exception" do
         expect { settings.get_setting(canonical_setting_name).validate_value }
-          .to raise_exception(ArgumentError, a_string_including("Both `#{canonical_setting_name}` and its deprecated alias `#{deprecated_setting_name}` have been set. Please only set `#{canonical_setting_name}`"))
+          .to raise_exception(java.lang.IllegalStateException, a_string_including("Both `#{canonical_setting_name}` and its deprecated alias `#{deprecated_setting_name}` have been set. Please only set `#{canonical_setting_name}`"))
       end
     end
   end

   context 'Settings#get on deprecated alias' do
     it 'produces a WARN-level message to the logger' do
-      expect(LogStash::Settings.logger).to receive(:warn).with(a_string_including "setting `#{canonical_setting_name}` has been queried by its deprecated alias `#{deprecated_setting_name}`")
       settings.get(deprecated_setting_name)
+      expect(events[0]).to include("setting `#{canonical_setting_name}` has been queried by its deprecated alias `#{deprecated_setting_name}`")
     end
   end
 end