Fix Kafka rebalancing when opentracing is not installed

CHANGELOG.md

@@ -53,6 +53,14 @@ resumes the Keep a Changelog format.
 - `faust[aerospike]` installed nothing: `requirements/extras/aerospike.txt`
   shipped without the matching `BUNDLES` entry in `setup.py`, despite being
   advertised in the README. A new test guards both directions of that mapping.
+- Every Kafka rebalance failed when `opentracing` was not installed. The no-op
+  stand-in Faust falls back to gave its spans no tracer, but
+  `traced_from_parent_span` starts a child span from `parent.tracer`, so
+  `on_partitions_revoked` and `on_partitions_assigned` both raised
+  `AttributeError`. Because `on_rebalance_start()` had already run, the app was
+  left mid-rebalance rather than crashing outright, surfacing as agents that
+  timed out and stalled. The stand-in now matches the real library across the
+  surface Faust uses, and a new parity test guards it.
 
 ### Changed
 - The `examples/fastapi/` directory is now `examples/fastapi_project/`. The old

faust/utils/_opentracing.py

@@ -6,7 +6,10 @@
 does real work when an ``app.tracer`` is configured (or the ``TracingSensor``
 is used), which requires the real ``opentracing`` package to be installed.
 
-This intentionally implements only the small surface Faust touches.
+Only the surface Faust touches is implemented, but across it this must stay
+substitutable: callers reach through what they are handed (``parent.tracer``),
+and absences count too -- the consumer thread reads ``operation_name`` and
+treats ``AttributeError`` as "not a real span".
 """
 
 from typing import Any, Literal
@@ -15,16 +18,15 @@
 class Span:
     """A span that does nothing."""
 
-    operation_name: str = ""
-
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
+    def __init__(self, tracer: Any, *args: Any, **kwargs: Any) -> None:
         self.context = _SpanContext()
-        self.tracer: Any = None
+        self.tracer: Any = tracer
 
     def __enter__(self) -> "Span":
         return self
 
     def __exit__(self, *exc_info: Any) -> Literal[False]:
+        self.finish()
         return False
 
     def finish(self, *args: Any, **kwargs: Any) -> None: ...
@@ -54,13 +56,10 @@ class Tracer:
     """A tracer that produces only no-op spans."""
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
-        self._noop_span = Span()
+        self._noop_span = Span(tracer=self)
 
     def start_span(self, *args: Any, **kwargs: Any) -> Span:
-        return Span()
-
-    def start_active_span(self, *args: Any, **kwargs: Any) -> Span:
-        return Span()
+        return self._noop_span
 
     def extract(self, *args: Any, **kwargs: Any) -> Any:
         return None
@@ -80,8 +79,10 @@ def child_of(*args: Any, **kwargs: Any) -> Any:
     return None
 
 
-def start_child_span(*args: Any, **kwargs: Any) -> Span:
-    return Span()
+def start_child_span(
+    parent_span: Span, operation_name: Any = None, *args: Any, **kwargs: Any
+) -> Span:
+    return parent_span.tracer.start_span(operation_name=operation_name)
 
 
 class Format:

tests/unit/utils/test_opentracing_shim.py

@@ -0,0 +1,168 @@
+"""Parity tests for the no-op :pypi:`opentracing` stand-in.
+
+Three kinds of check, none sufficient alone: surface comparison catches a name
+appearing or disappearing, value comparison catches a mirrored constant
+drifting, and functional calls catch a name that exists but holds the wrong
+value -- the bug that broke rebalancing, where ``Span.tracer`` existed and was
+``None``.
+
+These exercise the shim's objects directly, so they do not prove the
+``except ImportError`` fallback wiring itself.
+"""
+
+import opentracing
+import pytest
+from opentracing.ext import tags as real_tags
+
+from faust.utils import _opentracing as shim
+from faust.utils.tracing import current_span, set_current_span, traced_from_parent_span
+
+SPAN_SURFACE = frozenset(
+    {
+        "tracer",
+        "context",
+        "finish",
+        "set_tag",
+        "set_operation_name",
+        "log_kv",
+        "__enter__",
+        "__exit__",
+    }
+)
+
+SPAN_ABSENT = frozenset({"operation_name"})
+
+# ``start_active_span`` is excluded: the shim omits it by design.
+TRACER_SURFACE = frozenset({"start_span", "extract", "inject", "_noop_span"})
+
+# ``tags`` is excluded: the real one lives at ``opentracing.ext.tags``.
+MODULE_SURFACE = frozenset(
+    {
+        "tracer",
+        "follows_from",
+        "child_of",
+        "start_child_span",
+        "Format",
+        "Span",
+        "Tracer",
+    }
+)
+
+MODULES = [pytest.param(opentracing, id="real"), pytest.param(shim, id="shim")]
+
+SURFACES = [
+    pytest.param(
+        lambda m: m.Tracer().start_span(operation_name="p"), SPAN_SURFACE, id="span"
+    ),
+    pytest.param(lambda m: m.Tracer(), TRACER_SURFACE, id="tracer"),
+    pytest.param(lambda m: m, MODULE_SURFACE, id="module"),
+]
+
+CONSTANTS = [
+    pytest.param(
+        real_tags,
+        shim.tags,
+        (
+            "ERROR",
+            "SAMPLING_PRIORITY",
+            "SPAN_KIND",
+            "COMPONENT",
+            "MESSAGE_BUS_DESTINATION",
+        ),
+        id="tags",
+    ),
+    pytest.param(
+        opentracing.Format,
+        shim.Format,
+        ("TEXT_MAP", "HTTP_HEADERS", "BINARY"),
+        id="format",
+    ),
+]
+
+
+def noop_span(module):
+    return module.Tracer().start_span(operation_name="parity-probe")
+
+
+@pytest.fixture(autouse=True)
+def restore_current_span():
+    yield
+    set_current_span(None)
+
+
+@pytest.mark.parametrize("get, names", SURFACES)
+def test_surface_matches_real_library(get, names):
+    real, fake = get(opentracing), get(shim)
+    for name in names:
+        assert hasattr(real, name) == hasattr(fake, name), name
+
+
+@pytest.mark.parametrize("real, fake, names", CONSTANTS)
+def test_constant_values_match_real_library(real, fake, names):
+    for name in names:
+        assert getattr(real, name) == getattr(fake, name), name
+
+
+@pytest.mark.parametrize("module", MODULES)
+def test_start_span_returns_the_tracers_own_usable_noop_span(module):
+    tracer = module.Tracer()
+    span = tracer.start_span(operation_name="x")
+    assert span is tracer._noop_span
+    assert span.tracer.start_span(operation_name="child") is not None
+
+
+@pytest.mark.parametrize("module", MODULES)
+def test_span_hides_operation_name(module):
+    for name in SPAN_ABSENT:
+        assert not hasattr(noop_span(module), name)
+
+
+@pytest.mark.parametrize("module", MODULES)
+def test_start_child_span_returns_a_span_with_a_tracer(module):
+    assert module.start_child_span(noop_span(module), "child-op").tracer is not None
+
+
+@pytest.mark.parametrize("module", MODULES)
+def test_traced_from_parent_span_runs_with_a_noop_parent(module):
+    """Both rebalance callbacks hand this decorator a no-op span."""
+
+    @traced_from_parent_span(noop_span(module))
+    def work(value):
+        return value * 2
+
+    assert work(21) == 42
+
+
+@pytest.mark.parametrize("module", MODULES)
+async def test_traced_from_parent_span_awaits_a_coroutine(module):
+    """Production awaits, routing through ``corowrapped`` and ``_restore_span``."""
+    parent = noop_span(module)
+
+    @traced_from_parent_span(parent)
+    async def work(value):
+        return value * 2
+
+    assert await work(21) == 42
+    assert current_span() is parent
+
+
+def test_shim_exit_finishes_the_span():
+    finished = []
+
+    class RecordingSpan(shim.Span):
+        def finish(self, *args, **kwargs):
+            finished.append(True)
+
+    with RecordingSpan(tracer=shim.Tracer()):
+        pass
+
+    assert finished == [True]
+
+
+def test_shim_span_requires_a_tracer():
+    with pytest.raises(TypeError):
+        shim.Span()
+
+
+def test_shim_hides_start_active_span():
+    assert not hasattr(shim.Tracer, "start_active_span")