MassTransit/MassTransit · error · ArgumentNullException

connector

Error message

connector

What it means

The ConnectConsumeAuditObserver extension throws ArgumentNullException (parameter 'connector') when the IConsumeObserverConnector it is called on is null. The connector is the receiver of the audit observer, so it must be a live bus/massTransit instance.

Solutions

1. Ensure the bus is created and started before connecting the audit observer
2. Null-check the bus instance before wiring auditing
3. Resolve IBus from the DI container after host startup rather than a cached field

Example fix

// before
_bus.ConnectConsumeAuditObserver(store); // _bus may be null
// after
if (_bus == null) throw new InvalidOperationException("bus not started");
_bus.ConnectConsumeAuditObserver(store);

if (bus is null) throw new InvalidOperationException("bus must be started before connecting the audit observer");

bool CanAudit(IBus? bus) => bus is not null;

try { bus.ConnectConsumeAuditObserver(store); } catch (ArgumentNullException ex) { logger.LogError(ex, "null connector for audit observer"); throw; }

When it happens

Trigger: Invoking the extension on a null bus or null IBus instance, e.g. bus = await Bus.Factory... failing earlier or a field never initialized, then calling bus.ConnectConsumeAuditObserver(...).

Common situations: Audit wiring run before the bus is started/created; DI returning a null IBus; calling the extension on a stale disposed bus variable.

Related errors

• store
• Value cannot be null. (Parameter 'values')
• values
• pipe
• values

AI-assisted analysis of MassTransit/MassTransit@62ab339afa (2026-09-13). Data as JSON: /api/errors/f64fffdc063c8145. Report an issue: GitHub.

Appendix: source

Thrown at src/MassTransit/Configuration/AuditConfigurationExtensions.cs:46

            var sendHandle = connector.ConnectSendObserver(new AuditSendObserver(store, factory, specification.Filter));
            var publishHandle = connector.ConnectPublishObserver(new AuditPublishObserver(store, factory, specification.Filter));

            return new MultipleConnectHandle(sendHandle, publishHandle);
        }

        /// <summary>
        /// Add an observer that will audit consumed messages, sending them to the message audit store prior to consumption by the consumer
        /// </summary>
        /// <param name="connector">The bus or endpoint</param>
        /// <param name="store">The audit store</param>
        /// <param name="configureFilter">Filter configuration delegate</param>
        /// <param name="metadataFactory">Message metadata factory. If omitted, the default one will be used.</param>
        public static ConnectHandle ConnectConsumeAuditObserver(this IConsumeObserverConnector connector, IMessageAuditStore store,
            Action<IMessageFilterConfigurator> configureFilter = null, IConsumeMetadataFactory metadataFactory = null)
        {
            if (connector == null)
                throw new ArgumentNullException(nameof(connector));
            if (store == null)
                throw new ArgumentNullException(nameof(store));

            var filterConfigurator = new ConsumeMessageFilterConfigurator();
            configureFilter?.Invoke(filterConfigurator);

            var factory = metadataFactory ?? new DefaultConsumeMetadataFactory();

            return connector.ConnectConsumeObserver(new AuditConsumeObserver(store, factory, filterConfigurator.Filter));
        }
    }
}

View on GitHub (pinned to 62ab339afa)