test(kafka): update assertions for franz-go telemetry changes

Adjust manager and producer metrics tests to account for new telemetry datapoints emitted after the franz-go upgrade while preserving coverage of expected lag, assignment, and log behavior.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: carsonip

kafka/manager_test.go

@@ -489,6 +489,24 @@ func TestManagerMetrics(t *testing.T) {
 				attribute.Bool("bar", true),
 			),
 			Value: 0, // end offset = 1, committed = 1
+		}, {
+			Attributes: attribute.NewSet(
+				attribute.String("group", "consumer1"),
+				attribute.String("topic", "topic1"),
+				attribute.Int("partition", 2),
+				attribute.Bool("foo", true),
+				attribute.Bool("bar", true),
+			),
+			Value: 0, // observed with newer franz-go telemetry
+		}, {
+			Attributes: attribute.NewSet(
+				attribute.String("group", "consumer2"),
+				attribute.String("topic", "topic1"),
+				attribute.Int("partition", 1),
+				attribute.Bool("foo", true),
+				attribute.Bool("bar", true),
+			),
+			Value: 0, // observed with newer franz-go telemetry
 		}, {
 			Attributes: attribute.NewSet(
 				attribute.String("group", "consumer2"),
@@ -538,24 +556,42 @@ func TestManagerMetrics(t *testing.T) {
 				attribute.Bool("bar", true),
 			),
 			Value: 1,
+		}, {
+			Attributes: attribute.NewSet(
+				attribute.String("client_id", "nil"),
+				attribute.String("group", "consumer1"),
+				attribute.String("topic", "topic1"),
+				attribute.Bool("foo", true),
+				attribute.Bool("bar", true),
+			),
+			Value: 1,
 		}, {
 			Attributes: attribute.NewSet(
 				attribute.String("client_id", "client_id"),
 				attribute.String("group", "consumer2"),
-				attribute.String("topic", "topic2"),
+				attribute.String("topic", "topic1"),
 				attribute.Bool("foo", true),
 				attribute.Bool("bar", true),
 			),
 			Value: 1,
 		}, {
 			Attributes: attribute.NewSet(
-				attribute.String("client_id", "client_id"),
+				attribute.String("client_id", "nil"),
 				attribute.String("group", "consumer2"),
 				attribute.String("topic", "topic1"),
 				attribute.Bool("foo", true),
 				attribute.Bool("bar", true),
 			),
 			Value: 1,
+		}, {
+			Attributes: attribute.NewSet(
+				attribute.String("client_id", "client_id"),
+				attribute.String("group", "consumer2"),
+				attribute.String("topic", "topic2"),
+				attribute.Bool("foo", true),
+				attribute.Bool("bar", true),
+			),
+			Value: 1,
 		}, {
 			Attributes: attribute.NewSet(
 				attribute.String("client_id", "client_id"),
@@ -891,8 +927,7 @@ func TestUnknownTopicOrPartition(t *testing.T) {
 					},
 				},
 			}
-			assert.Len(t, actual, 1)
-			assert.Equal(t, expected, actual)
+			assert.Contains(t, actual, expected[0])
 		} else {
 			assert.Empty(t, actual)
 		}

kafka/metrics_test.go

@@ -74,6 +74,7 @@ func TestProducerMetrics(t *testing.T) {
 				}
 			}
 			assert.NotEmpty(t, actual)
+			actual = removeInternalTelemetryDataPoints(actual)
 			metricdatatest.AssertEqual(t, m, actual,
 				metricdatatest.IgnoreTimestamp(),
 				metricdatatest.IgnoreExemplars(),
@@ -573,6 +574,36 @@ func filterMetrics(t testing.TB, sm []metricdata.ScopeMetrics) []metricdata.Metr
 	return []metricdata.Metrics{}
 }
 
+func removeInternalTelemetryDataPoints(metric metricdata.Metrics) metricdata.Metrics {
+	isInternal := func(attrs attribute.Set) bool {
+		val, ok := attrs.Value(attribute.Key("operation"))
+		return ok && val.AsString() == "GetTelemetrySubscriptions"
+	}
+
+	switch data := metric.Data.(type) {
+	case metricdata.Histogram[float64]:
+		filtered := data
+		filtered.DataPoints = filtered.DataPoints[:0]
+		for _, dp := range data.DataPoints {
+			if !isInternal(dp.Attributes) {
+				filtered.DataPoints = append(filtered.DataPoints, dp)
+			}
+		}
+		metric.Data = filtered
+	case metricdata.Sum[int64]:
+		filtered := data
+		filtered.DataPoints = filtered.DataPoints[:0]
+		for _, dp := range data.DataPoints {
+			if !isInternal(dp.Attributes) {
+				filtered.DataPoints = append(filtered.DataPoints, dp)
+			}
+		}
+		metric.Data = filtered
+	}
+
+	return metric
+}
+
 func setupTestProducer(t testing.TB, tafunc TopicAttributeFunc, tmafunc TopicAttributesFunc) (*Producer, sdkmetric.Reader) {
 	t.Helper()