Add flush_after_seconds option to streaming_bulk()

[miguelgrinberg]

This change adds a timeout controlled flush to the streaming bulk helper.

Fixes #3051

Tasks:

• flush sentinel object
• flush unit tests
• flush integration tests
• timeout async support
• timeout sync support
• timeout integration tests
• docstring updates

[miguelgrinberg]

This assert may look strange. The reason I'm validating flushing by looking at timestamp differences is to attempt to not use any fixed timings and avoid a flaky test.

[pquentin]

Thanks! LGTM.

[pquentin]

Out of curiosity, do you have an example traceback that we get with this?

[miguelgrinberg]

Example test script:

import time
from elasticsearch.compat import safe_thread

def my_thread():
    print("my_thread started")
    1/0

def main():
    print("main started")
    with safe_thread(target=my_thread):
        time.sleep(1)

main()

Output:

❯ python threadtest.py
main started
my_thread started
Traceback (most recent call last):
  File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/threadtest.py", line 14, in <module>
    main()
    ~~~~^^
  File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/threadtest.py", line 10, in main
    with safe_thread(target=my_thread):
         ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^
  File "/Users/miguelgrinberg/.local/share/mise/installs/python/3.14.0/lib/python3.14/contextlib.py", line 148, in __exit__
    next(self.gen)
    ~~~~^^^^^^^^^^
  File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/elasticsearch/compat.py", line 105, in safe_thread
    raise captured_exception
  File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/elasticsearch/compat.py", line 95, in run
    target(*args, **kwargs)
    ~~~~~~^^^^^^^^^^^^^^^^^
  File "/Users/miguelgrinberg/Documents/dev/elasticsearch-py/s.py", line 6, in my_thread
    1/0
    ~^~
ZeroDivisionError: division by zero