Fix kafka flakiness (#8629)

## 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.

Author: NachoEchevarria

tracer/test/test-applications/integrations/Samples.Kafka/Producer.cs

@@ -89,7 +89,22 @@ private static int GetTopicPartitionCount(string topic, ClientConfig config)
 
         private static Partition GetPartition(ClientConfig config, string topic)
         {
-            var numPartitions = TopicPartitions.GetOrAdd(topic, t => GetTopicPartitionCount(topic, config));
+            // Only cache the partition count once metadata reports a non-zero
+            // value, otherwise a transient propagation lag would pin every
+            // produce on this topic to partition 0 for the rest of the run.
+            if (!TopicPartitions.TryGetValue(topic, out var numPartitions) || numPartitions <= 0)
+            {
+                numPartitions = GetTopicPartitionCount(topic, config);
+                if (numPartitions > 0)
+                {
+                    TopicPartitions[topic] = numPartitions;
+                }
+                else
+                {
+                    return new Partition(0);
+                }
+            }
+
             var partition = LastUsedPartition.GetOrAdd(topic, t => 0);
             if (partition >= numPartitions)
             {

tracer/test/test-applications/integrations/Samples.Kafka/TopicHelpers.cs

@@ -20,6 +20,7 @@ public static async Task<bool> TryCreateTopic(
             ClientConfig config)
         {
             var attempts = 3;
+            bool? created = null;
             do
             {
                 using var adminClient = new AdminClientBuilder(config).Build();
@@ -37,20 +38,63 @@ await adminClient.CreateTopicsAsync(new List<TopicSpecification> {
                     });
 
                     Console.WriteLine($"Topic created");
-                    return true;
+                    created = true;
+                    break;
                 }
                 catch (CreateTopicsException e)
                 {
                     if (e.Results[0].Error.Code == ErrorCode.TopicAlreadyExists)
                     {
                         Console.WriteLine("Topic already exists");
-                        return false;
+                        created = false;
+                        break;
                     }
 
                     Console.WriteLine($"An error occured creating topic {topicName}: {e.Results[0].Error.Reason}");
                 }
             } while (!TopicExists(topicName, config) && attempts-- > 0);
-            throw new Exception("Unable to create topic");
+
+            if (created is null)
+            {
+                throw new Exception("Unable to create topic");
+            }
+
+            // The create call returns as soon as the broker acks the request, but
+            // metadata propagation can lag behind. Wait until the topic is visible
+            // with the expected partition count, otherwise downstream callers that
+            // immediately query metadata (e.g. to pick a partition) can race and
+            // see a topic with zero partitions.
+            if (!await WaitForTopicMetadata(topicName, numPartitions, config, TimeSpan.FromSeconds(30)))
+            {
+                // Broker metadata propagation timed out — that's an infrastructure
+                // problem, not a tracer bug. Exit with the well-known skip code (13)
+                // so the test harness skips rather than fails this run.
+                Console.WriteLine($"Topic {topicName} metadata did not propagate with {numPartitions} partitions");
+                Console.WriteLine("Exiting with skip code (13)");
+                Environment.Exit(13);
+            }
+
+            return created.Value;
+        }
+
+        private static async Task<bool> WaitForTopicMetadata(string topicName, int expectedPartitions, ClientConfig config, TimeSpan timeout)
+        {
+            var deadline = DateTime.UtcNow + timeout;
+            while (DateTime.UtcNow < deadline)
+            {
+                using var adminClient = new AdminClientBuilder(config).Build();
+                var metadata = adminClient.GetMetadata(topicName, TimeSpan.FromSeconds(5));
+                var topic = metadata.Topics.FirstOrDefault(t => string.Equals(t.Topic, topicName, StringComparison.Ordinal));
+                if (topic != null && topic.Error.Code == ErrorCode.NoError && topic.Partitions.Count == expectedPartitions)
+                {
+                    return true;
+                }
+
+                Console.WriteLine($"Waiting for topic {topicName} metadata (partitions: {topic?.Partitions.Count ?? 0}/{expectedPartitions})...");
+                await Task.Delay(TimeSpan.FromMilliseconds(500));
+            }
+
+            return false;
         }
 
         /// <summary>