@task(task_run_name=f"process_batch", persist_result=False)
async def processBatch(project, dataset, schema, bucket_name, year, folder, batch, batch_number, total_batches):
    logger = get_run_logger()
    data_batch = []
    # Define a semaphore with a limit (e.g., 5 concurrent tasks)
    semaphore = Semaphore(20)
    
    async with aiohttp.ClientSession() as session:
        async with await get_storage_client(session) as storage:
            for count, gcs_file_url in enumerate(batch):
                # logger.debug(f"Processing file: {count+1} of {len(batch)} in batch {batch_number} of {total_batches} in {folder}")
                xpathFields, xpaths_hash = get_fields()
                metadata = await getMetadata(storage, gcs_file_url, bucket_name)
                if metadata['metadata']:
                    return_type = metadata['metadata']['RETURN_TYPE']
                    xpathFields.update(get_specific_fields(return_type))
                    # logger.debug(f"XPath fields retrieved for {gcsFileUrl}")
                
            # Process files concurrently with a limit
            data_batch = await asyncio.gather(
                *[getDataWithSemaphore(storage=storage, xpath_fields=xpath_fields, gcs_file_Url=gcs_file_url, bucket_name=bucket_name, semaphore=semaphore) for gcs_file_url in batch]
            )
    return data_batch