Repository: kedro-org/kedro
File: kedro/runner/task.py (Python)

Code before:
            try:
                inputs[name] = await input_task
            except Exception as exception:  # pragma: no cover
                raise exception

        is_async = True
        additional_inputs = self._collect_inputs_from_hook(
            node, catalog, inputs.copy(), is_async, hook_manager, session_id=session_id
        )
        inputs.update(additional_inputs)

        outputs = self._call_node_run(
            node, catalog, inputs, is_async, hook_manager, session_id=session_id
        )

        output_tasks = {
            name: asyncio.create_task(
                self._async_dataset_save(name, data, node, catalog, hook_manager)
            )
            for name, data in outputs.items()
        }

        for output_task in asyncio.as_completed(output_tasks.values()):
            try:
                await output_task
            except Exception as exception:
                raise exception

        return node

    @staticmethod
    async def _async_dataset_load(
        dataset_name: str,
        node: Node,
        catalog: CatalogProtocol,
        hook_manager: PluginManager,
    ) -> Any:
        """Asynchronously loads a dataset while ensuring hooks are executed in order."""
        hook_manager.hook.before_dataset_loaded(dataset_name=dataset_name, node=node)
        return_ds = await asyncio.to_thread(catalog.load, dataset_name)
        hook_manager.hook.after_dataset_loaded(
            dataset_name=dataset_name, data=return_ds, node=node
        )
        return return_ds

    @staticmethod
    async def _async_dataset_save(
        dataset_name: str,
        data: Any,
        node: Node,
        catalog: CatalogProtocol,

Review comment (bug):
Same question here for the exception. Or at least could we make the message below more specific to help with debugging?

Code after:
        # Gather results as they complete
        inputs = {}
        for name, input_task in input_tasks.items():
            try:
                inputs[name] = await input_task
            except Exception as exc:  # pragma: no cover
                logger.error(
                    "Error loading input '%s' for node '%s': %s",
                    name,
                    node.name,
                    exc,
                    exc_info=True,
                )
                raise

        is_async = True
        additional_inputs = self._collect_inputs_from_hook(
            node, catalog, inputs.copy(), is_async, hook_manager, session_id=session_id
        )
        inputs.update(additional_inputs)

        outputs = self._call_node_run(
            node, catalog, inputs, is_async, hook_manager, session_id=session_id
        )

        output_tasks = {
            name: asyncio.create_task(
                self._async_dataset_save(name, data, node, catalog, hook_manager)
            )
            for name, data in outputs.items()
        }

        for name, output_task in output_tasks.items():
            try:
                await output_task
            except Exception as exc:
                logger.error(
                    "Error saving output '%s' for node '%s': %s",
                    name,
                    node.name,
                    exc,
                    exc_info=True,
                )
                raise

        return node

    @staticmethod
    async def _async_dataset_load(
        dataset_name: str,
        node: Node,