def METHOD_NAME(
    self, tp: TP, offset: int, stream: StreamT, event: EventT
) -> None:
    stream_meta = getattr(event.message, "stream_meta", None)
    if stream_meta is None:
        stream_meta = event.message.stream_meta = {}  # type: ignore
    parent_span = event.message.span  # type: ignore
    if parent_span:
        stream_span = opentracing.start_child_span(
            parent_span, f"job-{event.message.topic}"
        )
        stream_span.set_tag("stream-concurrency-index", stream.concurrency_index)
        stream_span.set_tag("stream-prefix", stream.prefix)
        spans = stream_meta.get("stream_spans")
        if spans is None:
            spans = stream_meta["stream_spans"] = {}
        spans[stream] = stream_span
        stream_span.__enter__()