test_out_exec_filter: add a sleep to ensure the stream is emitted

[Watson1978]

Which issue(s) this PR fixes:
Fixes #4754

What this PR does / why we need it:
This PR will stable the tests.

Docs Changes:

Release Note:

[daipom]

Thanks for this fix!
The use of the driver in this test is wrong because Fluent::Test::Driver::BaseOwner keeps @event_streams.
expect_emits should be as follows.

d.run(default_tag: 'test', expect_emits: 1, ...
d.run(default_tag: 'test', expect_emits: 2, ...
d.run(default_tag: 'test', expect_emits: 3, ...
d.run(default_tag: 'test', expect_emits: 4, ...

The following would avoid the need to manage start and shutdown.
Could you try the following?

diff --git a/test/plugin/test_out_exec_filter.rb b/test/plugin/test_out_exec_filter.rb
index 5a7dc710..c808eda0 100644
--- a/test/plugin/test_out_exec_filter.rb
+++ b/test/plugin/test_out_exec_filter.rb
@@ -500,10 +500,15 @@ class ExecFilterOutputTest < Test::Unit::TestCase
     d = create_driver(conf)
     time = event_time('2011-01-02 13:14:15')
 
-    d.run(default_tag: 'test', expect_emits: 1, timeout: 10, start: true,  shutdown: false){ d.feed(time, {"k1" => 0}) }
-    d.run(default_tag: 'test', expect_emits: 1, timeout: 10, start: false, shutdown: false){ d.feed(time, {"k1" => 1}) }
-    d.run(default_tag: 'test', expect_emits: 1, timeout: 10, start: false, shutdown: false){ d.feed(time, {"k1" => 2}) }
-    d.run(default_tag: 'test', expect_emits: 1, timeout: 10, start: false, shutdown: false){ d.feed(time, {"k1" => 3}) }
+    d.run(default_tag: 'test', expect_emits: 4) do
+      d.feed(time, {"k1" => 0})
+      d.flush
+      d.feed(time, {"k1" => 1})
+      d.flush
+      d.feed(time, {"k1" => 2})
+      d.flush
+      d.feed(time, {"k1" => 3})
+    end
 
     assert_equal "2011-01-02 13:14:15\ttest\t0\n", d.formatted[0]
     assert_equal "2011-01-02 13:14:15\ttest\t1\n", d.formatted[1]
@@ -524,9 +529,6 @@ class ExecFilterOutputTest < Test::Unit::TestCase
     assert_equal pid_list[1], events[1][2]['child_pid']
     assert_equal pid_list[0], events[2][2]['child_pid']
     assert_equal pid_list[1], events[3][2]['child_pid']
-
-  ensure
-    d.run(start: false, shutdown: true)
   end
 
   # child process exits per 3 lines

[Watson1978]

@daipom I tried your suggession. However, unfortunatry it causes failure as following...

ExecFilterOutputTest:
  test: using child processes by round robin[with sections]:                                                                    F
====================================================================================================================================================
Failure: test: using child processes by round robin[with sections](ExecFilterOutputTest)
test/plugin/test_out_exec_filter.rb:530:in `block in <class:ExecFilterOutputTest>'
     527:
     528:     assert_equal pid_list[0], events[0][2]['child_pid']
     529:     assert_equal pid_list[1], events[1][2]['child_pid']
  => 530:     assert_equal pid_list[0], events[2][2]['child_pid']
     531:     assert_equal pid_list[1], events[3][2]['child_pid']
     532:   end
     533:
<"8132"> expected but was
<"11304">

diff:
? 8132
? 1  04
? ?  ?
====================================================================================================================================================
: (4.781393)

Finished in 4.7823206 seconds.
----------------------------------------------------------------------------------------------------------------------------------------------------
1 tests, 9 assertions, 1 failures, 0 errors, 0 pendings, 0 omissions, 0 notifications
0% passed
----------------------------------------------------------------------------------------------------------------------------------------------------
0.21 tests/s, 1.88 assertions/s
rake aborted!
Command failed with status (1): [bundle exec ruby -w -v -I"lib:test" -Eascii-8bit:ascii-8bit test/plugin/test_out_exec_filter.rb -v --name="test: using child processes by round robin[with sections]"]
C:/src/fluentd/runner.rake:4:in `block (2 levels) in <top (required)>'
C:/src/fluentd/runner.rake:2:in `times'
C:/src/fluentd/runner.rake:2:in `block in <top (required)>'
Tasks: TOP => default
(See full trace by running task with --trace)

[daipom]

@daipom I tried your suggession. However, unfortunatry it causes failure as following...

Hmm, I can't reproduce it.
Is it unstable?

Ubuntu focal, Ruby 3.2.2

  data(
    'with sections' => CONFIG_ROUND_ROBIN,
    'traditional' => CONFIG_ROUND_ROBIN_COMPAT,
  )
  test 'using child processes by round robin' do |conf|
    d = create_driver(conf)
    time = event_time('2011-01-02 13:14:15')

    d.run(default_tag: 'test', expect_emits: 4) do
      d.feed(time, {"k1" => 0})
      d.flush
      d.feed(time, {"k1" => 1})
      d.flush
      d.feed(time, {"k1" => 2})
      d.flush
      d.feed(time, {"k1" => 3})
    end

    assert_equal "2011-01-02 13:14:15\ttest\t0\n", d.formatted[0]
    assert_equal "2011-01-02 13:14:15\ttest\t1\n", d.formatted[1]
    assert_equal "2011-01-02 13:14:15\ttest\t2\n", d.formatted[2]
    assert_equal "2011-01-02 13:14:15\ttest\t3\n", d.formatted[3]

    events = d.events
    assert_equal 4, events.length

    pid_list = []
    events.each do |event|
      pid = event[2]['child_pid']
      pid_list << pid unless pid_list.include?(pid)
    end
    assert_equal 2, pid_list.size, "the number of pids should be same with number of child processes: #{pid_list.inspect}"

    assert_equal pid_list[0], events[0][2]['child_pid']
    assert_equal pid_list[1], events[1][2]['child_pid']
    assert_equal pid_list[0], events[2][2]['child_pid']
    assert_equal pid_list[1], events[3][2]['child_pid']
  end

[Watson1978]

Is it unstable?

In my environment, it fails with 5 ~ 10 %.
It should be repeated at least 20 times to confirm the results.

I tryed on Windows on virtual machine when I got the error.

[daipom]

Thanks!
Indeed, it sometimes fails on Windows.
Looks like we need to add some sleep to wait Output#write completion.

    d.run(default_tag: 'test', expect_emits: 4) do
      d.feed(time, {"k1" => 0})
      d.flush
      sleep 0.5
      d.feed(time, {"k1" => 1})
      d.flush
      sleep 0.5
      d.feed(time, {"k1" => 2})
      d.flush
      sleep 0.5
      d.feed(time, {"k1" => 3})
    end

[Watson1978]

Thanks. I updated this PR with your suggession.

[daipom]

Thanks!! Waiting CI...

Note: The reason why we need sleep here is as follows.

Fluent::Test::Driver::Output#wait_flush_completion does not accurately wait flush completion.

fluentd/lib/fluent/test/driver/output.rb

Lines 68 to 81 in 46372dd

	def wait_flush_completion
	buffer_queue = ->(){ @instance.buffer && @instance.buffer.queue.size > 0 }
	dequeued_chunks = ->(){
	@instance.dequeued_chunks_mutex &&
	@instance.dequeued_chunks &&
	@instance.dequeued_chunks_mutex.synchronize{ @instance.dequeued_chunks.size > 0 }
	}
	Timeout.timeout(10) do
	while buffer_queue.call || dequeued_chunks.call
	sleep 0.1
	end
	end
	end

It waits for buffer.queue to become empty.
However, dequeue happens at the beginning of the flush process.

fluentd/lib/fluent/plugin/output.rb

Lines 1188 to 1189 in 46372dd

	def try_flush
	chunk = @buffer.dequeue_chunk

So, it does not wait flush completion.

This test depends on the order of the flushes. Therefore, this makes the test unstable.
In the future, we should improve Fluent::Test::Driver::Output#wait_flush_completion instead of using sleep.

[daipom]

LGTM. Thanks!