code: def test_spillbuffer_fail_to_serialize(tmp_path):
      buf = SpillBuffer(str(tmp_path), target=200, max_spill=600)
  
      # bad data individually larger than spill threshold target 200
      a = Bad(size=201)
  
      # Exception caught in the worker
      with pytest.raises(TypeError, match="Failed to pickle 'a'") as e:
          with captured_logger("distributed.spill") as logs_bad_key:
              buf["a"] = a
      assert isinstance(e.value.__cause__.__cause__, MyError)
  
      # spill.py must remain silent because we're already logging in worker.py
      assert not logs_bad_key.getvalue()
      assert_buf(buf, tmp_path, {}, {})
  
      b = Bad(size=100)  # this is small enough to fit in memory/fast
  
      buf["b"] = b
      assert_buf(buf, tmp_path, {"b": b}, {})
  
      c = "c" * 100
      with captured_logger("distributed.spill") as logs_bad_key_mem:
          # This will go to fast and try to kick b out,
          # but keep b in fast since it's not pickable
          buf["c"] = c
  
      # worker.py won't intercept the exception here, so spill.py must dump the traceback
      logs_value = logs_bad_key_mem.getvalue()
      assert (
          "Failed to pickle 'b'" in logs_value
      )  # from distributed.spill
      assert "Traceback" in logs_value  # from distributed.spill
      assert_buf(buf, tmp_path, {"b": b, "c": c}, {})
  
imports: from __future__ import annotations
  import array
  import os
  import random
  import uuid
  from pathlib import Path
  import pytest
  import dask.config
  from dask.sizeof import sizeof
  from distributed import profile
  from distributed.compatibility import WINDOWS
  from distributed.metrics import meter
  from distributed.spill import SpillBuffer
  from distributed.utils import RateLimiterFilter
  from distributed.utils_test import captured_logger
label_desc: Use image node_id 7 for calling a global function with example usage: test_spillbuffer_fail_to_serialize(tmp_path) without return types