Fix the exception behavior when config.string contains ${VAR} in th…

…e comments. (elastic#16050)

* Wipe out comment lines if config comment contains.

* Remove substitution var process when loading the YAML, instead align on the generic approach which LSCL happens during the pipeline compile.

* Update logstash-core/src/main/java/org/logstash/config/ir/PipelineConfig.java

Put the logging config back as it is being used with composed configs.

logstash-core/lib/logstash/config/source/multi_local.rb

@@ -30,7 +30,7 @@ def initialize(settings)
     end
 
     def pipeline_configs
-      pipelines = deep_replace(retrieve_yaml_pipelines())
+      pipelines = retrieve_yaml_pipelines
       pipelines_settings = pipelines.map do |pipeline_settings|
         clone = @original_settings.clone
         clone.merge_pipeline_settings(pipeline_settings)

logstash-core/lib/logstash/util/substitution_variables.rb

@@ -36,7 +36,6 @@ def deep_replace(value)
       end
     else
       if value.is_a?(Array)
-        value_array_index = 0
         value.each_with_index do |single_value, i|
           value[i] = deep_replace(single_value)
         end

logstash-core/spec/logstash/config/source/multi_local_spec.rb

@@ -145,21 +145,33 @@
   end
 
   describe "#pipeline_configs" do
+
+    let(:config_string) {
+     "input {
+        udp {
+          port => 5555 # intentional comment contains \"${UDP_DEV_PORT}\" variable, shouldn't break functionalities
+          host => \"127.0.0.1\"
+        }
+        # another intentional comment contains \"${UDP_PROD_HOST}\" variable, shouldn't break functionalities
+      }
+      output {}"
+    }
     let(:retrieved_pipelines) do
       [
-        { "pipeline.id" => "main", "config.string" => "input {} output {}" },
-        { "pipeline.id" => "backup", "config.string" => "input {} output {}" }
+        { "pipeline.id" => "main", "config.string" => config_string },
+        { "pipeline.id" => "backup", "config.string" => config_string }
       ]
     end
+
     before(:each) do
       allow(subject).to receive(:retrieve_yaml_pipelines).and_return(retrieved_pipelines)
     end
 
     it "should return instances of PipelineConfig" do
       configs = subject.pipeline_configs
       expect(configs).to be_a(Array)
-      expect(subject.pipeline_configs.first).to be_a(::LogStash::Config::PipelineConfig)
-      expect(subject.pipeline_configs.last).to be_a(::LogStash::Config::PipelineConfig)
+      expect(configs.first).to be_a(::LogStash::Config::PipelineConfig)
+      expect(configs.last).to be_a(::LogStash::Config::PipelineConfig)
     end
 
     context "using non pipeline related settings" do