Limit the thread pool to 64 workers

Author: aabmass

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_upload/completion_hook.py

@@ -122,7 +122,9 @@ def __init__(
 
         # Use a ThreadPoolExecutor for its queueing and thread management. The semaphore
         # limits the number of queued tasks. If the queue is full, data will be dropped.
-        self._executor = ThreadPoolExecutor(max_workers=self._max_queue_size)
+        self._executor = ThreadPoolExecutor(
+            max_workers=min(self._max_queue_size, 64)
+        )
         self._semaphore = threading.BoundedSemaphore(self._max_queue_size)
 
     def _submit_all(self, upload_data: UploadData) -> None:

util/opentelemetry-util-genai/tests/test_upload.py

@@ -285,6 +285,19 @@ def test_upload_after_shutdown_logs(self):
             logs.output[0],
         )
 
+    def test_threadpool_max_workers(self):
+        for max_queue_size, expect_threadpool_workers in ((10, 10), (100, 64)):
+            with patch(
+                "opentelemetry.util.genai._upload.completion_hook.ThreadPoolExecutor"
+            ) as mock:
+                hook = UploadCompletionHook(
+                    base_path=BASE_PATH, max_queue_size=max_queue_size
+                )
+                self.addCleanup(hook.shutdown)
+                mock.assert_called_once_with(
+                    max_workers=expect_threadpool_workers
+                )
+
 
 class TestUploadCompletionHookIntegration(TestBase):
     def setUp(self):