rspec tests of clone and drop filters are broken on logstash-devutils 2.0.4

Hello! I'm using logstash-devutils to test my pipelines with rspec. I've just tried to update to version 2.0.4 (my current version 1.3.6 works too slow) and it stoped to correctly work with filters that delete or create new events (clone and drop). Here is how I reproduce it:

git clone git@github.com:elastic/logstash.git
cd logstash
sed -i 's/"logstash-devutils", "~> 1"/"logstash-devutils", "~> 2"/g' Gemfile.template
rake bootstrap
rake plugin:install-default
bin/logstash-plugin install --development
bin/rspec logstash-core/spec/logstash/custom_pipeline_spec.rb

where custom_pipeline_spec.rb is cutted copy of pipeline_spec.rb:

require "spec_helper"
require "logstash/inputs/generator"
require "logstash/filters/drop"
require_relative "../support/mocks_classes"
require_relative "../support/helpers"
require "stud/try"
require 'timeout'
require 'logstash/config/pipeline_config'

describe LogStash::Pipeline do
  let(:worker_thread_count)     { 5 }
  let(:safe_thread_count)       { 1 }
  let(:override_thread_count)   { 42 }
  let(:dead_letter_queue_enabled) { false }
  let(:dead_letter_queue_path) { }
  let(:pipeline_settings_obj) { LogStash::SETTINGS.clone }
  let(:pipeline_settings) { {} }
  let(:max_retry) {10} #times
  let(:timeout) {120} #seconds

  before :each do
    pipeline_workers_setting = LogStash::SETTINGS.get_setting("pipeline.workers")
    allow(pipeline_workers_setting).to receive(:default).and_return(worker_thread_count)
    dlq_enabled_setting = LogStash::SETTINGS.get_setting("dead_letter_queue.enable")
    allow(dlq_enabled_setting).to receive(:value).and_return(dead_letter_queue_enabled)
    dlq_path_setting = LogStash::SETTINGS.get_setting("path.dead_letter_queue")
    allow(dlq_path_setting).to receive(:value).and_return(dead_letter_queue_path)

    pipeline_settings.each {|k, v| pipeline_settings_obj.set(k, v) }
  end

  context "compiled flush function" do

    context "cancelled events should not propagate down the filters" do
      config <<-CONFIG
        filter {
          drop {}
        }
      CONFIG

      sample("hello") do
        expect(subject).to eq(nil)
      end
    end

    context "new events should propagate down the filters" do
      config <<-CONFIG
        filter {
          clone {
            clones => ["clone1"]
          }
        }
      CONFIG
      sample(["foo", "bar"]) do
        expect(subject.size).to eq(4)
      end
    end
  end

end

And result is:

Sending Logstash logs to null which is now configured via log4j2.properties
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/Users/IIIEII/Documents/Workspace/elk/logstash/logstash-core/lib/jars/log4j-slf4j-impl-2.13.3.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/Users/IIIEII/Documents/Workspace/elk/logstash/vendor/bundle/jruby/2.5.0/gems/march_hare-4.2.0-java/lib/ext/slf4j-simple.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
Run options: exclude {:integration=>true, :redis=>true, :socket=>true, :performance=>true, :couchdb=>true, :elasticsearch=>true, :elasticsearch_secure=>true, :export_cypher=>true, :windows=>true}

Randomized with seed 293
FF

Failures:

  1) LogStash::Pipeline compiled flush function new events should propagate down the filters "["foo","bar"]" when processed
     Failure/Error: expect(subject.size).to eq(4)

       expected: 4
            got: 2

       (compared using ==)
     # ./logstash-core/spec/logstash/pipeline_spec_2.rb:91:in `block in <main>'
     # ./spec/spec_helper.rb:66:in `block in <main>'
     # ./logstash-core/lib/logstash/util.rb:43:in `set_thread_name'
     # ./spec/spec_helper.rb:65:in `block in <main>'
     # ./spec/spec_helper.rb:58:in `block in <main>'
     # ./vendor/bundle/jruby/2.5.0/gems/logstash-devutils-2.0.4-java/lib/logstash/devutils/rspec/spec_helper.rb:51:in `block in <main>'
     # ./lib/bootstrap/rspec.rb:31:in `<main>'

  2) LogStash::Pipeline compiled flush function cancelled events should not propagate down the filters "hello" when processed
     Failure/Error: expect(subject).to eq(nil)

       expected: nil
            got: #<LogStash::Event:0x38eb2140>

       (compared using ==)
     # ./logstash-core/spec/logstash/pipeline_spec_2.rb:78:in `block in <main>'
     # ./spec/spec_helper.rb:66:in `block in <main>'
     # ./logstash-core/lib/logstash/util.rb:43:in `set_thread_name'
     # ./spec/spec_helper.rb:65:in `block in <main>'
     # ./spec/spec_helper.rb:58:in `block in <main>'
     # ./vendor/bundle/jruby/2.5.0/gems/logstash-devutils-2.0.4-java/lib/logstash/devutils/rspec/spec_helper.rb:51:in `block in <main>'
     # ./lib/bootstrap/rspec.rb:31:in `<main>'

Finished in 2.02 seconds (files took 2.82 seconds to load)
2 examples, 2 failures

Failed examples:

rspec ./logstash-core/spec/logstash/pipeline_spec_2.rb:90 # LogStash::Pipeline compiled flush function new events should propagate down the filters "["foo","bar"]" when processed
rspec ./logstash-core/spec/logstash/pipeline_spec_2.rb:77 # LogStash::Pipeline compiled flush function cancelled events should not propagate down the filters "hello" when processed

Randomized with seed 293

As you can see, drop and clone filters just do nothing. You have same number of event on output as on input.

elastic / logstash-devutils

rspec tests of clone and drop filters are broken on logstash-devutils 2.0.4 #90