def _threaded_download_and_preprocess_image(
        allocated_content: list[str],
        preprocessor,
        preprocessing_config: ImagePreprocessingConfig,
        metric_obj: Optional[RequestMetrics] = None,
        return_individual_error: bool = True,
) -> list[PreprocessedContent]:
    """A thread calls this function to download images for its allocated contents.

    Args:
        allocated_content: A list of URLs to be downloaded and preprocessed by this thread.
        preprocessor: The preprocessor to be used for preprocessing the image. E.g., OpenCLIPPreprocessor
        preprocessing_config: The preprocessing configuration to be used for preprocessing the image. This includes
            the image download timeout and the image download headers.
        return_individual_error: If True, collect individual errors in the thread_results, otherwise, raise an error.
        metric_obj: The telemetry object to be used for measuring the time taken for each thread.

    Ret
    """
    _id = f'media_download.{preprocessing_config.modality}.{threading.get_ident()}'
    thread_results: list[Union[InferenceErrorModel, list[Tuple[str, Tensor]]]] = []
    with metric_obj.time(f"{_id}.thread_time"):
        for url in allocated_content:
            try:
                image = load_image_from_path(
                    url, preprocessing_config.download_header, timeout_ms=preprocessing_config.download_timeout_ms,
                    metrics_obj=metric_obj
                )
            except PIL.UnidentifiedImageError as e:
                metric_obj.increment_counter(f"{url}.UnidentifiedImageError")
                if return_individual_error:
                    thread_results.append(InferenceErrorModel(error_message=str(e)))
                else:
                    raise MediaDownloadError(str(e))
                continue
            if isinstance(image, Image):
                try:
                    preprocessed_image: List[Tensor] = preprocessor.preprocess([image], preprocessing_config.modality)
                except OSError as e:
                    if "image file is truncated" in str(e):
                        if return_individual_error:
                            thread_results.append(InferenceErrorModel(error_message=f"Image file is truncated: {url}"))
                        else:
                            raise PreprocessingError(f"Image file is truncated: {url}")
                        continue
                    else:
                        raise e
                thread_results.append([(url, preprocessed_image[0])])
            else:
                if return_individual_error:
                    thread_results.append(InferenceErrorModel(error_message=f"Unexpected image type: {type(image)} "
                                                                            f"for image: {url}"))
                else:
                    raise ValueError(f"Unexpected image type: {type(image)} for image: {url}")
    return thread_results
A thread calls this function to download images for its allocated contents.

    Args:
        allocated_content: A list of URLs to be downloaded and preprocessed by this thread.
        preprocessor: The preprocessor to be used for preprocessing the image. E.g., OpenCLIPPreprocessor
        preprocessing_config: The preprocessing configuration to be used for preprocessing the image. This includes
            the image download timeout and the image download headers.
        return_individual_error: If True, collect individual errors in the thread_results, otherwise, raise an error.
        metric_obj: The telemetry object to be used for measuring the time taken for each thread.

    Ret
    
src/marqo/inference/media_download_and_preprocess/media_download_and_preprocess.py