diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java index a2d0b28799c..4032e966801 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/ConsumerCoordinatorInstrumentation.java @@ -10,6 +10,7 @@ import datadog.trace.api.datastreams.DataStreamsTags; import datadog.trace.bootstrap.InstrumentationContext; import datadog.trace.bootstrap.instrumentation.api.AgentTracer; +import datadog.trace.instrumentation.kafka_common.KafkaConfigHelper; import datadog.trace.instrumentation.kafka_common.MetadataState; import java.util.HashMap; import java.util.Map; @@ -61,6 +62,7 @@ public String instrumentedType() { public String[] helperClassNames() { return new String[] { packageName + ".KafkaConsumerInfo", + "datadog.trace.instrumentation.kafka_common.KafkaConfigHelper", "datadog.trace.instrumentation.kafka_common.PendingConfig", "datadog.trace.instrumentation.kafka_common.MetadataState", }; @@ -71,6 +73,9 @@ public void methodAdvice(MethodTransformer transformer) { transformer.applyAdvice( isMethod().and(named("sendOffsetCommitRequest")).and(takesArguments(1)), ConsumerCoordinatorInstrumentation.class.getName() + "$CommitOffsetAdvice"); + transformer.applyAdvice( + isMethod().and(named("onJoinComplete")).and(takesArguments(4)), + ConsumerCoordinatorInstrumentation.class.getName() + "$JoinGroupAdvice"); } public static class CommitOffsetAdvice { @@ -127,4 +132,46 @@ public static void muzzleCheck(ConsumerRecord record) { record.headers(); } } + + public static class JoinGroupAdvice { + @Advice.OnMethodExit(suppress = Throwable.class) + public static void trackJoinGroup( + @Advice.This ConsumerCoordinator coordinator, + @Advice.Argument(0) final int generationId, + @Advice.Argument(1) final String memberId, + @Advice.Argument(2) final String memberProtocol) { + if (memberId == null || memberId.isEmpty()) { + return; + } + KafkaConsumerInfo kafkaConsumerInfo = + InstrumentationContext.get(ConsumerCoordinator.class, KafkaConsumerInfo.class) + .get(coordinator); + if (kafkaConsumerInfo == null) { + return; + } + // Only report when the membership changes (new member id or new generation) to avoid + // re-reporting an unchanged membership. + if (memberId.equals(kafkaConsumerInfo.getLastReportedMemberId()) + && generationId == kafkaConsumerInfo.getLastReportedGenerationId()) { + return; + } + kafkaConsumerInfo.setLastReportedMembership(memberId, generationId); + + String consumerGroup = kafkaConsumerInfo.getConsumerGroup(); + Metadata consumerMetadata = kafkaConsumerInfo.getClientMetadata(); + String clusterId = null; + if (consumerMetadata != null) { + MetadataState metadataState = + InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata); + clusterId = metadataState != null ? metadataState.clusterId : null; + } + KafkaConfigHelper.reportConsumerGroupMember( + clusterId, consumerGroup, memberId, generationId, memberProtocol); + } + + public static void muzzleCheck(ConsumerRecord record) { + // Match CommitOffsetAdvice: only apply for kafka versions with headers + record.headers(); + } + } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java index 495847befe6..2065d9bb319 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInfo.java @@ -8,6 +8,10 @@ public class KafkaConsumerInfo { private final String consumerGroup; private final Metadata clientMetadata; private final String bootstrapServers; + // Last consumer group membership reported to DSM; used to avoid re-reporting when neither the + // member id nor the generation changed. Not part of identity (excluded from equals/hashCode). + private volatile String lastReportedMemberId; + private volatile int lastReportedGenerationId = Integer.MIN_VALUE; public KafkaConsumerInfo(String consumerGroup, Metadata clientMetadata, String bootstrapServers) { this.consumerGroup = consumerGroup; @@ -36,6 +40,20 @@ public String getBootstrapServers() { return bootstrapServers; } + @Nullable + public String getLastReportedMemberId() { + return lastReportedMemberId; + } + + public int getLastReportedGenerationId() { + return lastReportedGenerationId; + } + + public void setLastReportedMembership(String memberId, int generationId) { + this.lastReportedMemberId = memberId; + this.lastReportedGenerationId = generationId; + } + @Override public boolean equals(Object o) { if (this == o) return true; diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/ConsumerCoordinatorInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/ConsumerCoordinatorInstrumentation.java index 3bf531a703d..eaab363cbf4 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/ConsumerCoordinatorInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/ConsumerCoordinatorInstrumentation.java @@ -45,6 +45,7 @@ public String instrumentedType() { public String[] helperClassNames() { return new String[] { packageName + ".KafkaConsumerInfo", + "datadog.trace.instrumentation.kafka_common.KafkaConfigHelper", "datadog.trace.instrumentation.kafka_common.PendingConfig", "datadog.trace.instrumentation.kafka_common.MetadataState", }; @@ -55,5 +56,8 @@ public void methodAdvice(MethodTransformer transformer) { transformer.applyAdvice( isMethod().and(named("sendOffsetCommitRequest")).and(takesArguments(1)), packageName + ".ConsumerCoordinatorAdvice"); + transformer.applyAdvice( + isMethod().and(named("onJoinComplete")).and(takesArguments(4)), + packageName + ".JoinGroupAdvice"); } } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/JoinGroupAdvice.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/JoinGroupAdvice.java new file mode 100644 index 00000000000..60f7abd56a1 --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/JoinGroupAdvice.java @@ -0,0 +1,51 @@ +package datadog.trace.instrumentation.kafka_clients38; + +import datadog.trace.bootstrap.InstrumentationContext; +import datadog.trace.instrumentation.kafka_common.KafkaConfigHelper; +import datadog.trace.instrumentation.kafka_common.MetadataState; +import net.bytebuddy.asm.Advice; +import org.apache.kafka.clients.Metadata; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator; + +public class JoinGroupAdvice { + @Advice.OnMethodExit(suppress = Throwable.class) + public static void trackJoinGroup( + @Advice.This ConsumerCoordinator coordinator, + @Advice.Argument(0) final int generationId, + @Advice.Argument(1) final String memberId, + @Advice.Argument(2) final String memberProtocol) { + if (memberId == null || memberId.isEmpty()) { + return; + } + KafkaConsumerInfo kafkaConsumerInfo = + InstrumentationContext.get(ConsumerCoordinator.class, KafkaConsumerInfo.class) + .get(coordinator); + if (kafkaConsumerInfo == null) { + return; + } + // Only report when the membership changes (new member id or new generation) to avoid + // re-reporting an unchanged membership. + if (memberId.equals(kafkaConsumerInfo.getLastReportedMemberId().orElse(null)) + && generationId == kafkaConsumerInfo.getLastReportedGenerationId()) { + return; + } + kafkaConsumerInfo.setLastReportedMembership(memberId, generationId); + + String consumerGroup = kafkaConsumerInfo.getConsumerGroup().orElse(null); + Metadata consumerMetadata = kafkaConsumerInfo.getmetadata().orElse(null); + String clusterId = null; + if (consumerMetadata != null) { + MetadataState metadataState = + InstrumentationContext.get(Metadata.class, MetadataState.class).get(consumerMetadata); + clusterId = metadataState != null ? metadataState.clusterId : null; + } + KafkaConfigHelper.reportConsumerGroupMember( + clusterId, consumerGroup, memberId, generationId, memberProtocol); + } + + public static void muzzleCheck(ConsumerRecord record) { + // Match ConsumerCoordinatorAdvice: only apply for kafka versions with headers + record.headers(); + } +} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java index 947bbd623ae..8882996f6fe 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInfo.java @@ -8,6 +8,10 @@ public class KafkaConsumerInfo { private final String consumerGroup; private final Metadata metadata; private final String bootstrapServers; + // Last consumer group membership reported to DSM; used to avoid re-reporting when neither the + // member id nor the generation changed. Not part of identity (excluded from equals/hashCode). + private volatile String lastReportedMemberId; + private volatile int lastReportedGenerationId = Integer.MIN_VALUE; public KafkaConsumerInfo(String consumerGroup, Metadata metadata, String bootstrapServers) { this.consumerGroup = consumerGroup; @@ -33,6 +37,19 @@ public Optional getBootstrapServers() { return Optional.ofNullable(bootstrapServers); } + public Optional getLastReportedMemberId() { + return Optional.ofNullable(lastReportedMemberId); + } + + public int getLastReportedGenerationId() { + return lastReportedGenerationId; + } + + public void setLastReportedMembership(String memberId, int generationId) { + this.lastReportedMemberId = memberId; + this.lastReportedGenerationId = generationId; + } + @Override public boolean equals(Object o) { if (this == o) return true; diff --git a/dd-java-agent/instrumentation/kafka/kafka-common/src/main/java/datadog/trace/instrumentation/kafka_common/KafkaConfigHelper.java b/dd-java-agent/instrumentation/kafka/kafka-common/src/main/java/datadog/trace/instrumentation/kafka_common/KafkaConfigHelper.java index b9c5089f70a..bdaf9624453 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-common/src/main/java/datadog/trace/instrumentation/kafka_common/KafkaConfigHelper.java +++ b/dd-java-agent/instrumentation/kafka/kafka-common/src/main/java/datadog/trace/instrumentation/kafka_common/KafkaConfigHelper.java @@ -105,6 +105,31 @@ public static void storePendingConsumerConfig( log.debug("Stored pending consumer config (cluster ID not yet known)"); } + /** + * Reports consumer group membership when a consumer (re)joins a group. The broker-assigned member + * id is sent alongside the cluster id and consumer group. + */ + public static void reportConsumerGroupMember( + String clusterId, + String consumerGroup, + String memberId, + int generationId, + String memberProtocol) { + if (memberId == null || memberId.isEmpty()) { + return; + } + if (Config.get().isDataStreamsEnabled()) { + AgentTracer.get() + .getDataStreamsMonitoring() + .reportKafkaConsumerGroupMember( + clusterId != null ? clusterId : "", + consumerGroup != null ? consumerGroup : "", + memberId, + generationId, + memberProtocol != null ? memberProtocol : ""); + } + } + /** Called from metadata update advice when the cluster ID becomes available. */ public static void reportPendingConfig(MetadataState state, String clusterId) { PendingConfig pending = state.takePendingConfig(); diff --git a/dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java b/dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java index b6487c2c99d..13804a9f5d4 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java @@ -305,6 +305,26 @@ public void reportKafkaConfig( getThreadServiceName())); } + @Override + public void reportKafkaConsumerGroupMember( + String kafkaClusterId, + String consumerGroup, + String memberId, + int generationId, + String memberProtocol) { + inbox.offer( + new KafkaConfigReport( + "kafka_consumer", + kafkaClusterId, + consumerGroup, + memberId, + generationId, + memberProtocol, + Collections.emptyMap(), + timeSource.getCurrentTimeNanos(), + getThreadServiceName())); + } + @Override public void setCheckpoint(AgentSpan span, DataStreamsContext context) { PathwayContext pathwayContext = span.spanContext().getPathwayContext(); diff --git a/dd-trace-core/src/main/java/datadog/trace/core/datastreams/MsgPackDatastreamsPayloadWriter.java b/dd-trace-core/src/main/java/datadog/trace/core/datastreams/MsgPackDatastreamsPayloadWriter.java index 6b7c5ad32ea..ea9e530b01f 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/datastreams/MsgPackDatastreamsPayloadWriter.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/datastreams/MsgPackDatastreamsPayloadWriter.java @@ -57,6 +57,9 @@ public class MsgPackDatastreamsPayloadWriter implements DatastreamsPayloadWriter private static final byte[] CONFIG_TYPE = "Type".getBytes(ISO_8859_1); private static final byte[] CONFIG_KAFKA_CLUSTER_ID = "KafkaClusterId".getBytes(ISO_8859_1); private static final byte[] CONFIG_CONSUMER_GROUP = "ConsumerGroup".getBytes(ISO_8859_1); + private static final byte[] CONFIG_MEMBER_ID = "MemberId".getBytes(ISO_8859_1); + private static final byte[] CONFIG_GENERATION_ID = "GenerationId".getBytes(ISO_8859_1); + private static final byte[] CONFIG_MEMBER_PROTOCOL = "MemberProtocol".getBytes(ISO_8859_1); private static final byte[] CONFIG_ENTRIES = "Config".getBytes(ISO_8859_1); private static final int INITIAL_CAPACITY = 512 * 1024; @@ -290,7 +293,8 @@ private void writeKafkaConfigs(List configs, Writable packer) packer.writeUTF8(CONFIGS); packer.startArray(configs.size()); for (KafkaConfigReport config : configs) { - packer.startMap(4); // Type, KafkaClusterId, ConsumerGroup, Config + // Type, KafkaClusterId, ConsumerGroup, MemberId, GenerationId, MemberProtocol, Config + packer.startMap(7); packer.writeUTF8(CONFIG_TYPE); packer.writeString(config.getType(), null); @@ -301,6 +305,15 @@ private void writeKafkaConfigs(List configs, Writable packer) packer.writeUTF8(CONFIG_CONSUMER_GROUP); packer.writeString(config.getConsumerGroup(), null); + packer.writeUTF8(CONFIG_MEMBER_ID); + packer.writeString(config.getMemberId(), null); + + packer.writeUTF8(CONFIG_GENERATION_ID); + packer.writeLong(config.getGenerationId()); + + packer.writeUTF8(CONFIG_MEMBER_PROTOCOL); + packer.writeString(config.getMemberProtocol(), null); + packer.writeUTF8(CONFIG_ENTRIES); Map entries = config.getConfig(); packer.startMap(entries.size()); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DataStreamsWritingTest.java b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DataStreamsWritingTest.java index 35fef530c54..60aeeff01d9 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DataStreamsWritingTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DataStreamsWritingTest.java @@ -324,6 +324,57 @@ void writeKafkaConfigsToMockServer() throws InterruptedException, IOException { validateKafkaConfigMessage(requestBodies.get(0)); } + @Test + void writeKafkaConsumerGroupMemberToMockServer() throws InterruptedException, IOException { + WellKnownTags wellKnownTags = + new WellKnownTags( + "runtimeid", "hostname", "test", Config.get().getServiceName(), "version", "java"); + Config fakeConfig = mock(Config.class); + when(fakeConfig.getAgentUrl()).thenReturn(serverAddress.toString()); + when(fakeConfig.getWellKnownTags()).thenReturn(wellKnownTags); + when(fakeConfig.getPrimaryTag()).thenReturn("region-1"); + + OkHttpClient testOkhttpClient = OkHttpUtils.buildHttpClient(serverAddress, 5000L); + DDAgentFeaturesDiscovery features = mock(DDAgentFeaturesDiscovery.class); + when(features.supportsDataStreams()).thenReturn(true); + SharedCommunicationObjects sharedCommObjects = new SharedCommunicationObjects(); + sharedCommObjects.setFeaturesDiscovery(features); + sharedCommObjects.agentHttpClient = testOkhttpClient; + sharedCommObjects.createRemaining(fakeConfig); + + ControllableTimeSource timeSource = new ControllableTimeSource(); + TraceConfig traceConfig = mock(TraceConfig.class); + when(traceConfig.isDataStreamsEnabled()).thenReturn(true); + + DefaultDataStreamsMonitoring dataStreams = + new DefaultDataStreamsMonitoring( + fakeConfig, sharedCommObjects, timeSource, () -> traceConfig); + dataStreams.start(); + + // Report consumer group membership (as emitted on join) + dataStreams.reportKafkaConsumerGroupMember( + "cluster-1", "test-group", "consumer-1-abc123", 7, "range"); + + // Also add a stats point so the bucket has content + dataStreams.add( + new StatsPoint( + DataStreamsTags.create(null, null), + 9, + 0, + 10, + timeSource.getCurrentTimeNanos(), + 0, + 0, + 0, + null)); + + timeSource.advance(defaultBucketDurationNanos); + dataStreams.close(); + + awaitOneRequestBody(); + validateKafkaConsumerGroupMemberMessage(requestBodies.get(0)); + } + @Test void duplicateKafkaConfigsAreEachSerializedInPayload() throws InterruptedException, IOException { WellKnownTags wellKnownTags = @@ -407,13 +458,19 @@ private void validateKafkaConfigMessage(byte[] message) throws IOException { // Collect configs in a map keyed by type Map> configsByType = new HashMap<>(); for (int n = 0; n < numConfigs; n++) { - assertEquals(4, unpacker.unpackMapHeader()); + assertEquals(7, unpacker.unpackMapHeader()); assertEquals("Type", unpacker.unpackString()); String type = unpacker.unpackString(); assertEquals("KafkaClusterId", unpacker.unpackString()); unpacker.unpackString(); // skip cluster id value assertEquals("ConsumerGroup", unpacker.unpackString()); unpacker.unpackString(); // skip consumer group value + assertEquals("MemberId", unpacker.unpackString()); + unpacker.unpackString(); // skip member id value + assertEquals("GenerationId", unpacker.unpackString()); + unpacker.unpackLong(); // skip generation id value + assertEquals("MemberProtocol", unpacker.unpackString()); + unpacker.unpackString(); // skip member protocol value assertEquals("Config", unpacker.unpackString()); int configSize = unpacker.unpackMapHeader(); Map configEntries = new HashMap<>(); @@ -455,6 +512,61 @@ private void validateKafkaConfigMessage(byte[] message) throws IOException { assertTrue(foundStats, "Stats field not found in payload"); } + private void validateKafkaConsumerGroupMemberMessage(byte[] message) throws IOException { + GzipSource gzipSource = new GzipSource(Okio.source(new ByteArrayInputStream(message))); + BufferedSource bufferedSource = Okio.buffer(gzipSource); + MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(bufferedSource.inputStream()); + + int outerMapSize = unpacker.unpackMapHeader(); + boolean foundStats = false; + for (int i = 0; i < outerMapSize; i++) { + String key = unpacker.unpackString(); + if ("Stats".equals(key)) { + foundStats = true; + int numBuckets = unpacker.unpackArrayHeader(); + assertTrue(numBuckets >= 1); + + int bucketMapSize = unpacker.unpackMapHeader(); + boolean foundConfigs = false; + for (int j = 0; j < bucketMapSize; j++) { + String bucketKey = unpacker.unpackString(); + if ("Configs".equals(bucketKey)) { + foundConfigs = true; + int numConfigs = unpacker.unpackArrayHeader(); + assertEquals(1, numConfigs); + + assertEquals(7, unpacker.unpackMapHeader()); + assertEquals("Type", unpacker.unpackString()); + assertEquals("kafka_consumer", unpacker.unpackString()); + assertEquals("KafkaClusterId", unpacker.unpackString()); + assertEquals("cluster-1", unpacker.unpackString()); + assertEquals("ConsumerGroup", unpacker.unpackString()); + assertEquals("test-group", unpacker.unpackString()); + assertEquals("MemberId", unpacker.unpackString()); + assertEquals("consumer-1-abc123", unpacker.unpackString()); + assertEquals("GenerationId", unpacker.unpackString()); + assertEquals(7, unpacker.unpackLong()); + assertEquals("MemberProtocol", unpacker.unpackString()); + assertEquals("range", unpacker.unpackString()); + assertEquals("Config", unpacker.unpackString()); + // Membership reports carry no config entries + assertEquals(0, unpacker.unpackMapHeader()); + } else { + unpacker.skipValue(); + } + } + assertTrue(foundConfigs, "Configs field not found in bucket"); + + for (int b = 1; b < numBuckets; b++) { + unpacker.skipValue(); + } + } else { + unpacker.skipValue(); + } + } + assertTrue(foundStats, "Stats field not found in payload"); + } + private void validateDuplicateKafkaConfigMessage(byte[] message) throws IOException { GzipSource gzipSource = new GzipSource(Okio.source(new ByteArrayInputStream(message))); BufferedSource bufferedSource = Okio.buffer(gzipSource); @@ -481,13 +593,19 @@ private void validateDuplicateKafkaConfigMessage(byte[] message) throws IOExcept assertEquals(2, numConfigs); for (int n = 0; n < numConfigs; n++) { - assertEquals(4, unpacker.unpackMapHeader()); + assertEquals(7, unpacker.unpackMapHeader()); assertEquals("Type", unpacker.unpackString()); assertEquals("kafka_producer", unpacker.unpackString()); assertEquals("KafkaClusterId", unpacker.unpackString()); unpacker.unpackString(); // skip cluster id value assertEquals("ConsumerGroup", unpacker.unpackString()); unpacker.unpackString(); // skip consumer group value + assertEquals("MemberId", unpacker.unpackString()); + unpacker.unpackString(); // skip member id value + assertEquals("GenerationId", unpacker.unpackString()); + unpacker.unpackLong(); // skip generation id value + assertEquals("MemberProtocol", unpacker.unpackString()); + unpacker.unpackString(); // skip member protocol value assertEquals("Config", unpacker.unpackString()); int configSize = unpacker.unpackMapHeader(); Map configEntries = new HashMap<>(); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java index 6648e1a8c36..dbce0c580c8 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java @@ -1154,6 +1154,47 @@ void kafkaConsumerConfigIsReportedInBucket() throws InterruptedException { dataStreams.close(); } + @Test + void kafkaConsumerGroupMemberIsReportedInBucket() throws InterruptedException { + DDAgentFeaturesDiscovery features = stubFeatures(true); + ControllableTimeSource timeSource = new ControllableTimeSource(); + Sink sink = mock(Sink.class); + CapturingPayloadWriter payloadWriter = new CapturingPayloadWriter(); + TraceConfig traceConfig = stubTraceConfig(true); + + DefaultDataStreamsMonitoring dataStreams = + new DefaultDataStreamsMonitoring( + sink, + features, + timeSource, + () -> traceConfig, + payloadWriter, + DEFAULT_BUCKET_DURATION_NANOS); + dataStreams.start(); + dataStreams.reportKafkaConsumerGroupMember( + "cluster-1", "test-group", "consumer-1-abc123", 7, "range"); + timeSource.advance(DEFAULT_BUCKET_DURATION_NANOS); + dataStreams.report(); + + awaitBuckets(dataStreams, payloadWriter, 1); + + StatsBucket bucket = payloadWriter.buckets.get(0); + List kafkaConfigs = bucket.getKafkaConfigs(); + assertEquals(1, kafkaConfigs.size()); + KafkaConfigReport report = kafkaConfigs.get(0); + assertEquals("kafka_consumer", report.getType()); + assertEquals("cluster-1", report.getKafkaClusterId()); + assertEquals("test-group", report.getConsumerGroup()); + assertEquals("consumer-1-abc123", report.getMemberId()); + assertEquals(7, report.getGenerationId()); + assertEquals("range", report.getMemberProtocol()); + assertTrue(report.getConfig().isEmpty()); + + // cleanup + payloadWriter.close(); + dataStreams.close(); + } + @Test void duplicateKafkaConfigsAreEachReportedInTheBucket() throws InterruptedException { DDAgentFeaturesDiscovery features = stubFeatures(true); diff --git a/internal-api/src/main/java/datadog/trace/api/datastreams/AgentDataStreamsMonitoring.java b/internal-api/src/main/java/datadog/trace/api/datastreams/AgentDataStreamsMonitoring.java index 42654dfda7c..7ebc4e421c8 100644 --- a/internal-api/src/main/java/datadog/trace/api/datastreams/AgentDataStreamsMonitoring.java +++ b/internal-api/src/main/java/datadog/trace/api/datastreams/AgentDataStreamsMonitoring.java @@ -22,6 +22,24 @@ public interface AgentDataStreamsMonitoring void reportKafkaConfig( String type, String kafkaClusterId, String consumerGroup, Map config); + /** + * Reports Kafka consumer group membership for Data Streams Monitoring. Called every time a + * consumer (re)joins a group, carrying the broker-assigned member id alongside the cluster id and + * consumer group. + * + * @param kafkaClusterId the Kafka cluster identifier, or empty string if not yet known + * @param consumerGroup the consumer group name + * @param memberId the broker-assigned consumer group member id + * @param generationId the consumer group generation/epoch at join time + * @param memberProtocol the negotiated partition assignment protocol (e.g. "range") + */ + void reportKafkaConsumerGroupMember( + String kafkaClusterId, + String consumerGroup, + String memberId, + int generationId, + String memberProtocol); + /** * Tracks Schema Registry usage for Data Streams Monitoring. * diff --git a/internal-api/src/main/java/datadog/trace/api/datastreams/KafkaConfigReport.java b/internal-api/src/main/java/datadog/trace/api/datastreams/KafkaConfigReport.java index d940847e22f..de0b5950761 100644 --- a/internal-api/src/main/java/datadog/trace/api/datastreams/KafkaConfigReport.java +++ b/internal-api/src/main/java/datadog/trace/api/datastreams/KafkaConfigReport.java @@ -1,16 +1,28 @@ package datadog.trace.api.datastreams; +import java.util.Collections; import java.util.Map; import java.util.Objects; /** * KafkaConfigReport captures Kafka producer or consumer configuration to be sent via the Data * Streams payload. Each unique configuration is sent only once. + * + *

It is also reused to report consumer-group membership: when a consumer (re)joins a group, a + * report is emitted carrying the broker-assigned {@code memberId} (plus the group {@code + * generationId} and the negotiated {@code memberProtocol}) alongside the cluster id and consumer + * group, with an empty {@link #config} map. */ public class KafkaConfigReport implements InboxItem { + /** Sentinel generation id for reports that are not membership reports. */ + public static final int NO_GENERATION = -1; + private final String type; // "kafka_producer" or "kafka_consumer" private final String kafkaClusterId; private final String consumerGroup; + private final String memberId; // broker-assigned consumer group member id, or "" when unknown + private final int generationId; // consumer group generation/epoch, or NO_GENERATION + private final String memberProtocol; // negotiated partition assignment protocol, or "" private final Map config; private final long timestampNanos; private final String serviceNameOverride; @@ -22,10 +34,35 @@ public KafkaConfigReport( Map config, long timestampNanos, String serviceNameOverride) { + this( + type, + kafkaClusterId, + consumerGroup, + "", + NO_GENERATION, + "", + config, + timestampNanos, + serviceNameOverride); + } + + public KafkaConfigReport( + String type, + String kafkaClusterId, + String consumerGroup, + String memberId, + int generationId, + String memberProtocol, + Map config, + long timestampNanos, + String serviceNameOverride) { this.type = type; this.kafkaClusterId = kafkaClusterId != null ? kafkaClusterId : ""; this.consumerGroup = consumerGroup != null ? consumerGroup : ""; - this.config = config; + this.memberId = memberId != null ? memberId : ""; + this.generationId = generationId; + this.memberProtocol = memberProtocol != null ? memberProtocol : ""; + this.config = config != null ? config : Collections.emptyMap(); this.timestampNanos = timestampNanos; this.serviceNameOverride = serviceNameOverride; } @@ -42,6 +79,18 @@ public String getConsumerGroup() { return consumerGroup; } + public String getMemberId() { + return memberId; + } + + public int getGenerationId() { + return generationId; + } + + public String getMemberProtocol() { + return memberProtocol; + } + public Map getConfig() { return config; } @@ -59,9 +108,12 @@ public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; KafkaConfigReport that = (KafkaConfigReport) o; - return Objects.equals(type, that.type) + return generationId == that.generationId + && Objects.equals(type, that.type) && Objects.equals(kafkaClusterId, that.kafkaClusterId) && Objects.equals(consumerGroup, that.consumerGroup) + && Objects.equals(memberId, that.memberId) + && Objects.equals(memberProtocol, that.memberProtocol) && Objects.equals(config, that.config); } @@ -70,6 +122,9 @@ public int hashCode() { int result = type != null ? type.hashCode() : 0; result = 31 * result + (kafkaClusterId != null ? kafkaClusterId.hashCode() : 0); result = 31 * result + (consumerGroup != null ? consumerGroup.hashCode() : 0); + result = 31 * result + (memberId != null ? memberId.hashCode() : 0); + result = 31 * result + generationId; + result = 31 * result + (memberProtocol != null ? memberProtocol.hashCode() : 0); result = 31 * result + (config != null ? config.hashCode() : 0); return result; } diff --git a/internal-api/src/main/java/datadog/trace/api/datastreams/NoopDataStreamsMonitoring.java b/internal-api/src/main/java/datadog/trace/api/datastreams/NoopDataStreamsMonitoring.java index 09bd3ee1e3a..2e87be10dbb 100644 --- a/internal-api/src/main/java/datadog/trace/api/datastreams/NoopDataStreamsMonitoring.java +++ b/internal-api/src/main/java/datadog/trace/api/datastreams/NoopDataStreamsMonitoring.java @@ -70,6 +70,14 @@ public void trackTransaction( public void reportKafkaConfig( String type, String kafkaClusterId, String consumerGroup, Map config) {} + @Override + public void reportKafkaConsumerGroupMember( + String kafkaClusterId, + String consumerGroup, + String memberId, + int generationId, + String memberProtocol) {} + @Override public void setConsumeCheckpoint(String type, String source, DataStreamsContextCarrier carrier) {}