How to design polling with changing interval in Kotlin?

There is a "buffer of elements" that a program must process somehow with function fun process(Element element){ ... }. Most of the time (80%) the queue is empty, and other time its load is not predictable: maybe 1 element per minute, maybe 1000 elements per second. We cannot see the whole buffer population, just ask "give at most X elements for processing". The goal is continuously process elements as fast as we can, optimally utilizing current executor.

Executor could be anything with contract fun submit(delay: Duration, task: () -> T): Unit, where each task somehow leads to calling process(element).

By "optimally utilizing current executor" I mean:

1. submit minimal amount of tasks when the queue is empty
2. submit just enough tasks to process current load
3. avoid long running tasks (i.e. longer than 5 seconds)
4. do not overpopulate the executor's tasks queue

(see an example below)

What is a known approach to implement such a component (tasks source)?

Such things as Quartz or Timer seem to have no needed flexibility out of the box to be useful for solving the problem.

I tried several attempts to implement this "flexible scheduling", and found out that there are several complex corner cases and it will take significant amount of time to implement a correct solution. Thereby as a trade-off for now I decided to use a straightforward solution with constant interval polling, which is too slow on high load and too fussy on low load, but at least works stable.

Did not manage to find any relevant information on the topic; maybe I use misleading keywords. So anything would be helpful: at least links to articles or discussions. And it would be perfect if there is a production ready tool that solves this problem.

Example:

  The component continuously submits tasks in an infinite loop.
  Its reaction affects only amount of tasks per submission and their delay reported to the executor.
--------------------------------------------------------
   event                        component's reaction  
--------------------------------------------------------
 queue is empty              detected zero load
                                - decrease amount of tasks being submitted (to not less than 1)
                                - increase (i.e. double) delay for next task submission

 10 elements arrived         detected some load
                                - keep amount of tasks unchanged (no info about the load size)
                                - decrease delay to 0

 queue is empty              it appears that on task processed the full load at on turn
                                - keep amount of tasks unchanged
                                - increase delay for next task submission 
   
 1000 elements arrived       detected some load
                                - keep amount of tasks unchanged (no info about the load size)
                                - decrease delay to 0

 990 still in queue          still have some load
                                - increase (i.e. +1) amount of tasks
                                - keep delay on 0 

 ...

 500 still in queue          still have some load (... now we submit 10 tasks per cycle)
                                - keep amount of tasks unchanged 
                                  (because executor has some tasks 
                                   in "not started" and "not finished" states)
                                - keep delay on 0 
 
 ...

 queue is empty              detected zero load
                                - decrease amount of tasks being submitted (to not less than 1)
                                - increase (i.e. double) delay for next task submission