{"record":{"id":"41b4d3fd03e6e941","repo":"microsoft/aspire","slug":"invalid-consumer-type","errorCode":null,"errorMessage":"Invalid consumer type.","messagePattern":"Invalid consumer type\\.","errorType":"validation","errorClass":"ArgumentException","httpStatus":null,"severity":"error","filePath":"src/Vendoring/OpenTelemetry.Instrumentation.ConfluentKafka/OpenTelemetryConsumeResultExtensions.cs","lineNumber":89,"sourceCode":"    /// <typeparam name=\"TValue\">The type of value of the <see cref=\"ConsumeResult{TKey,TValue}\"/>.</typeparam>\n    /// <returns>A <see cref=\"ValueTask\"/>.</returns>\n    public static async ValueTask<ConsumeResult<TKey, TValue>?> ConsumeAndProcessMessageAsync<TKey, TValue>(\n        this IConsumer<TKey, TValue> consumer,\n        OpenTelemetryConsumeAndProcessMessageHandler<TKey, TValue> handler,\n        CancellationToken cancellationToken)\n    {\n#if NETFRAMEWORK\n        if (consumer == null)\n        {\n            throw new ArgumentNullException(nameof(consumer));\n        }\n#else\n        ArgumentNullException.ThrowIfNull(consumer);\n#endif\n\n        if (consumer is not InstrumentedConsumer<TKey, TValue> instrumentedConsumer)\n        {\n            throw new ArgumentException(\"Invalid consumer type.\", nameof(consumer));\n        }\n\n#if NETFRAMEWORK\n        if (handler is null)\n        {\n            throw new ArgumentNullException(nameof(handler));\n        }\n#else\n        ArgumentNullException.ThrowIfNull(handler);\n#endif\n\n        var consumeResult = instrumentedConsumer.Consume(cancellationToken);\n\n        if (consumeResult?.Message == null || consumeResult.IsPartitionEOF)\n        {\n            return consumeResult;\n        }\n","sourceCodeStart":71,"sourceCodeEnd":107,"githubUrl":"https://github.com/microsoft/aspire/blob/25830f84bd145686607ad00c057b3f84e2e51d43/src/Vendoring/OpenTelemetry.Instrumentation.ConfluentKafka/OpenTelemetryConsumeResultExtensions.cs#L71-L107","documentation":"ConsumeAndProcessMessageAsync wraps a Confluent Kafka consumer with OpenTelemetry instrumentation, but only its own InstrumentedConsumer<TKey,TValue> carries the required telemetry state. Passing any other IConsumer implementation makes instrumented processing impossible, so it throws ArgumentException naming the consumer parameter.","triggerScenarios":"Calling ConsumeAndProcessMessageAsync with a raw Confluent.Kafka.ConsumerBuilder-built consumer instead of the instrumented consumer created by the vendored OpenTelemetry Confluent.Kafka instrumentation.","commonSituations":"Mixing a plain Confluent.Kafka consumer (or a test double) with the instrumentation extension method; a refactor replaced the instrumented wrapper with a bare consumer; DI registered the wrong consumer type.","solutions":["Create the consumer via the instrumentation's builder/extension so it returns Instrumented<TKey,TValue>","Unwrap or cast to the instrumented consumer before calling the extension","Register the instrumented consumer type in DI instead of the raw IConsumer"],"exampleFix":"// before\nusing var consumer = new ConsumerBuilder<TKey, TValue>(config).Build();\nawait consumer.ConsumeAndProcessMessageAsync(handler, ct); // throws\n// after\nusing var consumer = new ConsumerBuilder<TKey, TValue>(config)\n    .BuildWithInstrumentation(); // returns InstrumentedConsumer<TKey, TValue>\nawait consumer.ConsumeAndProcessMessageAsync(handler, ct);","handlingStrategy":"type-guard","validationCode":"if (consumer is not InstrumentedConsumer<TKey, TValue>)\n    throw new InvalidOperationException(\"Use the instrumentation's BuildWithInstrumentation() consumer.\");","typeGuard":"static bool IsInstrumented<TK,TV>(IConsumer<TK,TV> c) => c is InstrumentedConsumer<TK,TV>;","tryCatchPattern":"try { await consumer.ConsumeAndProcessMessageAsync(handler, ct); }\ncatch (ArgumentException ex) { log.LogError(ex, \"Consumer must be the instrumented wrapper\"); }","preventionTips":["Always create consumers via the instrumentation builder extension","Keep DI registrations on InstrumentedConsumer<TKey,TValue>","Don't swap the instrumented wrapper for a raw Confluent.Kafka consumer in refactors"],"tags":["kafka","opentelemetry","consumer"],"backgroundTag":"type-mismatch","analyzedSha":"25830f84bd145686607ad00c057b3f84e2e51d43","analyzedAt":"2026-09-16T11:10:06.193Z","contentChangedAt":"2026-09-16T11:10:06.193Z","schemaVersion":2},"datasetVersion":"2026-09-21T09:17:21.228Z"}