Perf: avoid the rest of Thread.list calls

lib/logstash/plugin_mixins/jdbc/scheduler.rb

@@ -25,6 +25,29 @@ def timeout_jobs
       super
     end
 
+    # @overload
+    def work_threads(query = :all)
+      if query.nil? # our special case from JobDecorator#start_work_thread
+        @_work_threads = nil # when a new worker thread is being added reset
+        return super(:all)
+      end
+
+      # Gets executed every time a job is triggered, we're going to cache the
+      # worker threads for this scheduler (to avoid `Thread.list`) - they only
+      # change when a new thread is being started from #start_work_thread ...
+      work_threads = @_work_threads
+      if work_threads.nil?
+        work_threads = threads.select { |t| t[:rufus_scheduler_work_thread] }
+        @_work_threads = work_threads
+      end
+
+      case query
+      when :active then work_threads.select { |t| t[:rufus_scheduler_job] }
+      when :vacant then work_threads.reject { |t| t[:rufus_scheduler_job] }
+      else work_threads
+      end
+    end
+
     # @overload
     def on_error(job, err)
       details = { exception: err.class, message: err.message, backtrace: err.backtrace }
@@ -85,14 +108,14 @@ def do_schedule(job_type, t, callable, opts, return_job_instance, block)
     module JobDecorator
 
       def start_work_thread
-        prev_thread_count = @scheduler.work_threads.size
+        prev_thread_count = @scheduler.work_threads(nil).size
 
         ret = super() # does not return Thread instance in 3.0
 
-        work_threads = @scheduler.work_threads
+        work_threads = @scheduler.work_threads(nil)
         while prev_thread_count == work_threads.size # very unlikely
           Thread.pass
-          work_threads = @scheduler.work_threads
+          work_threads = @scheduler.work_threads(nil)
         end
 
         work_thread_name_prefix = @scheduler.work_thread_name_prefix