<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def parallel_bulk( client, actions, thread_count=4, chunk_size=500, max_chunk_bytes=100 * 1024 * 1024, queue_size=4, expand_action_callback=expand_action, *args, **kwargs ):
 """ Parallel version of the bulk helper run in multiple threads at once. :arg client: instance of :class:`~elasticsearch.Elasticsearch` to use :arg actions: iterator containing the actions :arg thread_count: size of the threadpool to use for the bulk requests :arg chunk_size: number of docs in one chunk sent to es (default: 500) :arg max_chunk_bytes: the maximum size of the request in bytes (default: 100MB) :arg raise_on_error: raise ``BulkIndexError`` containing errors (as `.errors`) from the execution of the last chunk when some occur. By default we raise. :arg raise_on_exception: if ``False`` then don't propagate exceptions from call to ``bulk`` and just report the items that failed as failed. :arg expand_action_callback: callback executed on each action passed in, should return a tuple containing the action line and the data line (`None` if data line should be omitted). :arg queue_size: size of the task queue between the main thread (producing chunks to send) and the processing threads. """

    # Avoid importing multiprocessing unless parallel_bulk is used
    # to avoid exceptions on restricted environments like App Engine
    from multiprocessing.pool import ThreadPool

    actions = map(expand_action_callback, actions)

    class BlockingPool(ThreadPool):
        def _setup_queues(self):
            super(BlockingPool, self)._setup_queues()
            # The queue must be at least the size of the number of threads to 
            # prevent hanging when inserting sentinel values during teardown.
            self._inqueue = Queue(max(queue_size, thread_count))
            self._quick_put = self._inqueue.put

    pool = BlockingPool(thread_count)

    try:
        for result in pool.imap(
            lambda bulk_chunk: list(
                _process_bulk_chunk(
                    client, bulk_chunk[1], bulk_chunk[0], *args, **kwargs
                )
            ),
            _chunk_actions(
                actions, chunk_size, max_chunk_bytes, client.transport.serializer
            ),
        ):
            for item in result:
                yield item

    finally:
        pool.close()
        pool.join()