fix flushing upon empty batches with ordered execution

when running a pipeline with ordered execution, flushes on the pipeline
were no longer being called when compute is called with an empty batch, causing
issues with the aggregate filter, for example, not being able to push events on
timeout.

logstash-core/src/main/java/org/logstash/config/ir/CompiledPipeline.java

@@ -306,23 +306,35 @@ private boolean isOutput(final Vertex vertex) {
 
     public final class CompiledOrderedExecution extends CompiledExecution {
 
+        @SuppressWarnings({"unchecked"})  private final RubyArray<RubyEvent> EMPTY_ARRAY = RubyUtil.RUBY.newEmptyArray();
+
         @Override
         public void compute(final QueueBatch batch, final boolean flush, final boolean shutdown) {
            compute(batch.events(), flush, shutdown);
         }
 
         @Override
         public void compute(final Collection<RubyEvent> batch, final boolean flush, final boolean shutdown) {
-            @SuppressWarnings({"unchecked"}) final RubyArray<RubyEvent> outputBatch = RubyUtil.RUBY.newArray();
-            // send batch one-by-one as single-element batches down the filters
-            @SuppressWarnings({"unchecked"}) final RubyArray<RubyEvent> filterBatch = RubyUtil.RUBY.newArray(1);
-            for (final RubyEvent e : batch) {
-                filterBatch.set(0, e);
-                final Collection<RubyEvent> result = compiledFilters.compute(filterBatch, flush, shutdown);
-                copyNonCancelledEvents(result, outputBatch);
-                compiledFilters.clear();
+            if (!batch.isEmpty()) {
+                @SuppressWarnings({"unchecked"}) final RubyArray<RubyEvent> outputBatch = RubyUtil.RUBY.newArray();
+                @SuppressWarnings({"unchecked"}) final RubyArray<RubyEvent> filterBatch = RubyUtil.RUBY.newArray(1);
+                // send batch one-by-one as single-element batches down the filters
+                for (final RubyEvent e : batch) {
+                    filterBatch.set(0, e);
+                    _compute(filterBatch, outputBatch, flush, shutdown);
+                }
+                compiledOutputs.compute(outputBatch, flush, shutdown);
+            } else if (flush || shutdown) {
+                @SuppressWarnings({"unchecked"}) final RubyArray<RubyEvent> outputBatch = RubyUtil.RUBY.newArray();
+                _compute(EMPTY_ARRAY, outputBatch, flush, shutdown);
+                compiledOutputs.compute(outputBatch, flush, shutdown);
             }
-            compiledOutputs.compute(outputBatch, flush, shutdown);
+        }
+
+        private void _compute(final RubyArray<RubyEvent> batch, final RubyArray<RubyEvent> outputBatch, final boolean flush, final boolean shutdown) {
+            final Collection<RubyEvent> result = compiledFilters.compute(batch, flush, shutdown);
+            copyNonCancelledEvents(result, outputBatch);
+            compiledFilters.clear();
         }
     }