microsoft/aspire · error · ArgumentException

Invalid consumer type.

Error message

Invalid consumer type.

What it means

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.

Solutions

1. Create the consumer via the instrumentation's builder/extension so it returns Instrumented<TKey,TValue>
2. Unwrap or cast to the instrumented consumer before calling the extension
3. Register the instrumented consumer type in DI instead of the raw IConsumer

Example fix

// before
using var consumer = new ConsumerBuilder<TKey, TValue>(config).Build();
await consumer.ConsumeAndProcessMessageAsync(handler, ct); // throws
// after
using var consumer = new ConsumerBuilder<TKey, TValue>(config)
    .BuildWithInstrumentation(); // returns InstrumentedConsumer<TKey, TValue>
await consumer.ConsumeAndProcessMessageAsync(handler, ct);

if (consumer is not InstrumentedConsumer<TKey, TValue>)
    throw new InvalidOperationException("Use the instrumentation's BuildWithInstrumentation() consumer.");

static bool IsInstrumented<TK,TV>(IConsumer<TK,TV> c) => c is InstrumentedConsumer<TK,TV>;

try { await consumer.ConsumeAndProcessMessageAsync(handler, ct); }
catch (ArgumentException ex) { log.LogError(ex, "Consumer must be the instrumented wrapper"); }

When it happens

Trigger: Calling ConsumeAndProcessMessageAsync with a raw Confluent.Kafka.ConsumerBuilder-built consumer instead of the instrumented consumer created by the vendored OpenTelemetry Confluent.Kafka instrumentation.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• Cannot cast ' ' from ' ' to
• Must be in the range
• Must not be null
• Must not be null or empty
• Must not be null or whitespace

AI-assisted analysis of microsoft/aspire@25830f84bd (2026-09-16). Data as JSON: /api/errors/41b4d3fd03e6e941. Report an issue: GitHub.

Appendix: source

Thrown at src/Vendoring/OpenTelemetry.Instrumentation.ConfluentKafka/OpenTelemetryConsumeResultExtensions.cs:89

    /// <typeparam name="TValue">The type of value of the <see cref="ConsumeResult{TKey,TValue}"/>.</typeparam>
    /// <returns>A <see cref="ValueTask"/>.</returns>
    public static async ValueTask<ConsumeResult<TKey, TValue>?> ConsumeAndProcessMessageAsync<TKey, TValue>(
        this IConsumer<TKey, TValue> consumer,
        OpenTelemetryConsumeAndProcessMessageHandler<TKey, TValue> handler,
        CancellationToken cancellationToken)
    {
#if NETFRAMEWORK
        if (consumer == null)
        {
            throw new ArgumentNullException(nameof(consumer));
        }
#else
        ArgumentNullException.ThrowIfNull(consumer);
#endif

        if (consumer is not InstrumentedConsumer<TKey, TValue> instrumentedConsumer)
        {
            throw new ArgumentException("Invalid consumer type.", nameof(consumer));
        }

#if NETFRAMEWORK
        if (handler is null)
        {
            throw new ArgumentNullException(nameof(handler));
        }
#else
        ArgumentNullException.ThrowIfNull(handler);
#endif

        var consumeResult = instrumentedConsumer.Consume(cancellationToken);

        if (consumeResult?.Message == null || consumeResult.IsPartitionEOF)
        {
            return consumeResult;
        }

View on GitHub (pinned to 25830f84bd)