code: def load_bytes(self, keys):
      tmpdir = mkdtemp(
          dir=self._tmproot, prefix="metaflow.gs.load_bytes."
      )
      try:
          futures = [
              self._executor.submit(
                  self.root_client.load_bytes_single,
                  tmpdir,
                  key,
              )
              for key in keys
          ]
  
          # Let's detect any failures fast. Stop and cleanup ASAP.
          # Note that messing up return order is beneficial re: Hyrum's law too.
          items = [future.result() for future in as_completed(futures)]
      except Exception:
          # Other BaseExceptions will skip clean up here.
          # We let this go - smooth exit in those circumstances is more important than cleaning up a few tmp files
          if os.path.exists(tmpdir):
              shutil.rmtree(tmpdir)
          raise
  
      class _Closer(object):
          @staticmethod
          def close():
              if os.path.isdir(tmpdir):
                  shutil.rmtree(tmpdir)
  
      return CloseAfterUse(iter(items), closer=_Closer)
  
imports: import json
  import os
  import shutil
  import uuid
  from concurrent.futures import as_completed
  from tempfile import mkdtemp
  from metaflow.datastore.datastore_storage import DataStoreStorage, CloseAfterUse
  from metaflow.exception import MetaflowInternalError
  from metaflow.metaflow_config import DATASTORE_SYSROOT_GS, ARTIFACT_LOCALROOT, GS_STORAGE_WORKLOAD_TYPE
  from metaflow.plugins.gcp.gs_storage_client_factory import get_gs_storage_client
  from metaflow.plugins.gcp.gs_utils import check_gs_deps, parse_gs_full_path, process_gs_exception
  from metaflow.plugins.storage_executor import StorageExecutor, handle_executor_exceptions
label_desc: Use image node_id 9 for calling the GSStorage obj's underlying member method code with example usage: obj.load_bytes(keys) and returns: CloseAfterUse