Add integration for goka (github.com/lovoo/goka) — APM tracing + Data Streams Monitoring

[anapsix]

Is your feature request related to a problem?

There is no dd-trace-go integration for goka, the Kafka-streams
library. goka builds its consumer group on top of IBM/sarama, but the existing
contrib/IBM/sarama integration cannot instrument a goka processor: goka's *Processor
is the sarama.ConsumerGroupHandler and passes itself directly to
consumerGroup.Consume(ctx, topics, g) (goka processor.go). Because there is no
user-supplied handler to wrap, the sarama contrib's handler-wrapping never attaches, and
consumed messages get no spans and no DSM checkpoints. Today users must hand-roll
instrumentation (we maintain two separate local patches — one for DSM, one for APM
propagation — that hook the same seam and duplicate a goka-header carrier).

Describe the solution you'd like

A new contrib/lovoo/goka integration (module
github.com/DataDog/dd-trace-go/contrib/lovoo/goka/v2, package goka) that provides both
APM distributed tracing and Data Streams Monitoring, built entirely on goka's existing public
extension points — no changes to goka are required:

• Consume side: goka.WithContextWrapper (per-input-message hook) to extract the
  propagated trace/DSM context from Context.Headers(), set the inbound DSM checkpoint
  (direction:in) and TrackKafkaCommitOffset, and start a kafka.consume span.
• Produce side: a wrapping goka.Context whose Emit/Loopback inject the active span
  and a direction:out DSM checkpoint into outbound headers via goka.WithCtxEmitHeaders.
• Helpers for the standalone goka.Emitter path and for extracting a span context, mirroring
  the twmb/franz-go and IBM/sarama contribs.

Proposed surface:

tr := gokatrace.NewTracer(gokatrace.WithService("orders"), gokatrace.WithDataStreams())

p, _ := goka.NewProcessor(brokers, goka.DefineGroup(group,
    goka.Input(topic, codec, tr.WrapCallback(handle)),
), goka.WithContextWrapper(tr.WrapContext))

The integration follows the established Kafka-family conventions: instrumentation.Load with
a new PackageLovooGoka, WithService/WithDataStreams options, the base64 DSM carrier,
kafka.consume/kafka.produce (v0) span names, the standard messaging tags
(span.kind=consumer, component=lovoo/goka, messaging.system=kafka,
messaging.destination.name, partition/offset), mocktracer + pathway-hash tests, and
replace => ../../.. in a dedicated submodule.

Notes

• Why a contrib and not a goka patch: goka carries no Datadog/OpenTelemetry dependency and its
  maintainers keep it telemetry-free; the required seams are already public and stable
  (WithContextWrapper since goka v1.1.8, WithCtxEmitHeaders since v1.0.6). Pinning target:
  goka v1.1.8 (minimum version exposing WithContextWrapper).
• No orchestrion.yml: unlike sarama/franz-go there is no single constructor call to rewrite
  (a processor is assembled from a user-defined graph plus options), so the integration is an
  explicit one-line opt-in rather than auto-instrumented.

Describe alternatives you've considered

• Patching goka to emit spans directly — rejected: adds a Datadog dependency the maintainers
  won't accept, and couples goka's release cadence to dd-trace-go.
• Injecting an instrumented sarama.Config/handler through goka's builder options
  (WithConsumerGroupBuilder, etc.) — does not help on the consume side because goka owns the
  handler; WithContextWrapper is the sanctioned hook.
• sarama KIP-42 interceptors (Consumer.Interceptors / Producer.Interceptors) injected via
  goka's config builders — this is the seam OpenTelemetry-style instrumentation uses, and a
  ConsumerInterceptor does fire for goka (its consumer group routes through the
  partitionConsumer responseFeeder, which applies interceptors before claim.Messages()).
  Rejected as the primary seam because interceptors are strictly weaker for a stream processor:
  (1) OnConsume fires before the callback with no "processing done" counterpart, so it cannot
  scope a span around the goka callback or tag errors from ctx.Fail; (2) an interceptor sees
  a raw *sarama.ConsumerMessage, not goka's Context, so it cannot propagate the consume
  span into goka Emit/Loopback. It could still complement WithContextWrapper for
  lower-level produce/consume spans if desired. Note dd-trace-go's IBM/sarama contrib
  currently ships no interceptor implementation.

I have a working implementation with unit and integration tests and am happy to open the PR
once the approach is agreed.