## Summary of changes

Fixes flake in DSM tests by using unique but stable topic names

> Supersedes #7209

## Reason for change

[Saw some flake in a test
recently](https://dev.azure.com/datadoghq/dd-trace-dotnet/_build/results?buildId=181163&view=logs&j=b7fc299e-b4c1-5ec0-1e5a-2167c3cc6270&t=2406fbf1-c048-53d0-bb5f-56f6b7adbd43):
`10:19:53 [DBG] Unhandled exception.
Confluent.Kafka.ProduceException`2[System.String,System.String]: Local:
Unknown topic`

What's more, the creation logs are kind of strange.

```
10:19:53 [DBG]  Creating topics...
10:19:53 [DBG]  Trying to delete topic data-streams-1...
10:19:53 [DBG]  Trying to create topic data-streams-1...
10:19:53 [DBG]  Topic already exists
10:19:53 [DBG]  Trying to delete topic data-streams-2...
10:19:53 [DBG]  Trying to delete topic data-streams-2...
10:19:53 [DBG]  Topic did not exist, skipping
10:19:53 [DBG]  Trying to create topic data-streams-2...
10:19:53 [DBG]  Topic created
10:19:53 [DBG]  Trying to delete topic data-streams-3...
10:19:53 [DBG]  Trying to create topic data-streams-3...
10:19:53 [DBG]  Topic created
10:19:53 [DBG]  Finished creating topics: 0:00:02.1158037
```

I can't really understand that sequence of logs, but deletes don't
happen atomically, so not entirely surprised by weirdness.

#7209 tried to work around it by using random topic names, but the
checkpoints are dependent on the topic names. This approach uses stable,
but unique, values

## Implementation details

Pass the topic name suffix to use to the sample from the test. This is
the approach we use in the existing non-DSM `KafkaTests` to avoid flake.

## Test coverage

The existing coverage is sufficient, but requires updating all the
snapshots.

## Other details

Supersedes #7209

## Summary of changes

Fix the `DataStreamsMonitoringKafkaTests.HandlesBatchProcessing` (and
related) flake by making the test sample wait for Kafka metadata to
propagate after topic creation, and by hardening the partition-picker in
the producer helper against a transient `0` partition count.

## Reason for change

The `HandlesBatchProcessing` test failed in CI with a Verify mismatch
where every `kafka_produce` backlog entry collapsed onto `partition:0`:

```
- partition:1, topic:data-streams-batch-processing-1-..., type:kafka_produce
- partition:1, topic:data-streams-batch-processing-2-..., type:kafka_produce
- partition:2, topic:data-streams-batch-processing-1-..., type:kafka_produce
- partition:2, topic:data-streams-batch-processing-2-..., type:kafka_produce
```

The four missing entries are all producer-side backlogs for partitions 1
and 2. Consumer-side `kafka_commit` backlogs are present for all
partitions, and the per-topic in/out DSM edges are present too — only
the producer collapsed onto partition 0.

**Direct confirmation in the sample's stdout** — every produce went to
partition `[[0]]`:

```
Produced message to: data-streams-batch-processing-1-HandlesBatchProcessing [[0]] @0
Produced message to: data-streams-batch-processing-1-HandlesBatchProcessing [[0]] @1
Produced message to: data-streams-batch-processing-1-HandlesBatchProcessing [[0]] @2
...
Produced message to: data-streams-batch-processing-2-HandlesBatchProcessing [[0]] @0
Produced message to: data-streams-batch-processing-2-HandlesBatchProcessing [[0]] @1
Produced message to: data-streams-batch-processing-2-HandlesBatchProcessing [[0]] @2
```

…instead of the expected `[[0]]`, `[[1]]`, `[[2]]` distribution. Topic
creation took 4.2s (`Finished creating topics: 0:00:04.2190968`) and
producing started ~1s later — well within the metadata-propagation
window where `AdminClient.GetMetadata` can return the topic with zero
partitions.

**Root cause** is a race in the test sample:

1. `TopicHelpers.TryCreateTopic` returns as soon as Kafka acks the
`CreateTopics` request, but broker metadata propagation can lag.
2. `Producer.GetPartition` calls `AdminClient.GetMetadata(topic, 5s)` to
learn the partition count. If metadata hasn't propagated yet, the topic
comes back with zero partitions and `GetTopicPartitionCount` returns
`0`.
3. That `0` gets cached in a static `ConcurrentDictionary<string,int>`
for the lifetime of the sample process via `GetOrAdd`.
4. With `numPartitions == 0`, the `partition >= numPartitions` guard is
always true, so every subsequent produce is pinned to partition 0 — for
the entire run.

This is timing-dependent (flaky, not always failing), and once it loses
the race the whole run is doomed.

This is the same family of flake addressed in #7211 ("Alternative
approach to fixing flaky DSM Kafka tests"), but the partition-count race
in the sample wasn't covered there.

## Implementation details

Two changes, both in the test sample only — no tracer code touched.

**`Samples.Kafka/TopicHelpers.cs` — fix the root cause.** After
`CreateTopicsAsync` succeeds (or returns `TopicAlreadyExists`), poll
`GetMetadata` until the topic is visible with the expected partition
count, with a 30s bounded timeout. If metadata never propagates, throw a
descriptive exception instead of silently returning. The original
throw-on-exhausted-retries behavior is preserved.

**`Samples.Kafka/Producer.cs` — defense in depth.** `GetPartition` no
longer caches a `0` partition count. If `GetTopicPartitionCount` returns
0 (shouldn't happen anymore after the TopicHelpers fix, but in case
anyone calls this without going through `TryCreateTopic`), we fall back
to partition 0 for that single call and re-query on the next produce
instead of pinning the whole run.

## Test coverage

Existing `DataStreamsMonitoringKafkaTests` cover the scenario; the
failure mode this PR fixes is exactly the one observed in the failing
build. No new tests added — this is a flake fix in a sample app, not a
behavior change in the tracer.

## Other details

- Test sample code only; no tracer / production code changes.
- `Samples.Kafka/TopicHelpers.cs` is used by every Kafka-producing
sample test, so it broadly hardens that family of tests against the
metadata-propagation race.