feat: Add support Apache Pulsar #3682

lillo42 · 2025-07-23T21:56:22Z

Add support to Apache Pulsar #3655

Apache Pulsar will create the Topic/Subscription if it doesn't exists
The current .NET implementation of Apache Pulsar doesn't have support to dead letter queue or retry queue

# Conflicts: # Brighter.sln

# Conflicts: # .github/workflows/ci.yml # Brighter.sln # Directory.Packages.props

# Conflicts: # .github/workflows/ci.yml

codescene-delta-analysis

Gates Failed
Enforce advisory code health rules (3 files with Constructor Over-Injection, Complex Method, Code Duplication)

Gates Passed
3 Quality Gates Passed

See analysis details in CodeScene

Reason for failure

Enforce advisory code health rules	Violations	Code Health Impact
PulsarMessageConsumer.cs	1 advisory rule	10.00 → 9.39	Suppress
PulsarSubscription.cs	1 advisory rule	10.00 → 9.69	Suppress
PulsarMessageProducer.cs	1 advisory rule	10.00 → 9.69	Suppress

Quality Gate Profile: Clean Code Collective
Want more control? Customize Code Health rules or catch issues early with our IDE extension and CLI tool.

codescene-delta-analysis · 2025-10-03T07:06:34Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarSubscription.cs

+    public PulsarSubscription(SubscriptionName subscriptionName, ChannelName channelName, RoutingKey routingKey, 
+        Type? requestType = null, Func<Message, Type>? getRequestType = null, int bufferSize = 1, int noOfPerformers = 1,
+        TimeSpan? timeOut = null, int requeueCount = -1, TimeSpan? requeueDelay = null, int unacceptableMessageLimit = 0,
+        MessagePumpType messagePumpType = MessagePumpType.Unknown, IAmAChannelFactory? channelFactory = null,
+        OnMissingChannel makeChannels = OnMissingChannel.Create, TimeSpan? emptyChannelDelay = null, 
+        TimeSpan? channelFailureDelay = null,
+        ISchema<ReadOnlySequence<byte>>? schema = null,
+        SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition.Earliest, 
+        int priorityLevel = 0, 
+        bool readCompacted = false, 
+        SubscriptionType subscriptionType = SubscriptionType.Exclusive, 
+        bool allowOutOfOrderDeliver = false,
+        Action<IConsumerBuilder<ReadOnlySequence<byte>>>? configuration = null) 
+        : base(subscriptionName, channelName, routingKey, requestType, getRequestType, bufferSize, noOfPerformers, timeOut, requeueCount, 
+            requeueDelay, unacceptableMessageLimit, messagePumpType, channelFactory, makeChannels, emptyChannelDelay, 
+            channelFailureDelay)
+    {
+        Schema = schema ?? DotPulsar.Schema.ByteSequence;
+        InitialPosition = initialPosition;
+        PriorityLevel = priorityLevel;
+        ReadCompacted = readCompacted;
+        SubscriptionType = subscriptionType;
+        AllowOutOfOrderDeliver = allowOutOfOrderDeliver;
+        Configuration = configuration;
+    }


❌ New issue: Constructor Over-Injection
PulsarSubscription has 23 arguments, max arguments = 5

_Suppress

codescene-delta-analysis · 2025-10-03T07:06:34Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarSubscription.cs

+    public PulsarSubscription(SubscriptionName subscriptionName, ChannelName channelName, RoutingKey routingKey,
+        Func<Message, Type>? getRequestType = null, int bufferSize = 1, int noOfPerformers = 1, TimeSpan? timeOut = null,
+        int requeueCount = -1, TimeSpan? requeueDelay = null, int unacceptableMessageLimit = 0,
+        MessagePumpType messagePumpType = MessagePumpType.Unknown, IAmAChannelFactory? channelFactory = null,
+        OnMissingChannel makeChannels = OnMissingChannel.Create, TimeSpan? emptyChannelDelay = null,
+        TimeSpan? channelFailureDelay = null,
+        ISchema<ReadOnlySequence<byte>>? schema = null,
+        SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition.Earliest,
+        int priorityLevel = 0,
+        bool readCompacted = false,
+        SubscriptionType subscriptionType = SubscriptionType.Exclusive,
+        bool allowOutOfOrderDeliver = false,
+        Action<IConsumerBuilder<ReadOnlySequence<byte>>>? configuration = null)
+        : base(subscriptionName, channelName, routingKey, typeof(T), getRequestType, bufferSize, noOfPerformers, timeOut, requeueCount,
+            requeueDelay, unacceptableMessageLimit, messagePumpType, channelFactory, makeChannels, emptyChannelDelay,
+            channelFailureDelay, schema, initialPosition, priorityLevel, readCompacted, subscriptionType, 
+            allowOutOfOrderDeliver, configuration)
+    {
+
+    }


❌ New issue: Constructor Over-Injection
PulsarSubscription has 22 arguments, max arguments = 5

_Suppress

iancooper

I am probably missing something here, but this feels like we are re-implementing our message pump inside the consumer, which doesn't feel like the "right thing" to be doing. Maybe we need a catch up conversation

iancooper · 2025-10-24T18:15:09Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumerFactory.cs

+            throw new ConfigurationException("We expect PulsarSubscription or PulsarSubscription<T> as a parameter");
+        }
+
+        var background = s_backgroundConsumers.GetOrAdd(pulsarSubscription, CreateConsumerBackground);


We might want to chat about what this is, as I don't know Pulsar? But I might need to understand it to see how this works with our pump

iancooper · 2025-10-24T18:16:13Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumerFactory.cs

+
+        var consumer = builder.Create();
+
+        return new PulsarBackgroundMessageConsumer(pulsarSubscription.NoOfPerformers, consumer);


We tend to be single threaded and rely on our performers to start multiple pumps. This would seen to counteract that, but, again I don't know Pulsar well enough to understand that.

iancooper · 2025-10-24T18:17:19Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarBackgroundMessageConsumer.cs

+    ///   <item><description>Subsequent calls increment the reference count but don't start additional loops</description></item>
+    /// </list>
+    /// </remarks>
+    public void Start()


Why do we have a background loop here? Why not just consume as we do elsewhere and rely on the single-threaded pump to scale performers?

iancooper · 2025-10-24T18:17:54Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarBackgroundMessageConsumer.cs

+        {
+            try
+            {
+                var message = await consumer.Receive(cancellationToken);


This feels like a pump within our pump?

# Conflicts: # Brighter.sln # src/Paramore.Brighter/Observability/MessagingSystem.cs

codescene-delta-analysis

Gates Failed
Enforce advisory code health rules (3 files with Constructor Over-Injection, Complex Method, Code Duplication)

Gates Passed
3 Quality Gates Passed

See analysis details in CodeScene

Reason for failure

Enforce advisory code health rules	Violations	Code Health Impact
PulsarMessageConsumer.cs	1 advisory rule	9.39	Suppress
PulsarSubscription.cs	1 advisory rule	9.69	Suppress
PulsarMessageProducer.cs	1 advisory rule	9.69	Suppress

Quality Gate Profile: Clean Code Collective
Want more control? Customize Code Health rules or catch issues early with our IDE extension and CLI tool.

codescene-delta-analysis · 2025-12-03T09:55:19Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarSubscription.cs

+    public PulsarSubscription(SubscriptionName subscriptionName, ChannelName channelName, RoutingKey routingKey, 
+        Type? requestType = null, Func<Message, Type>? getRequestType = null, int bufferSize = 1, int noOfPerformers = 1,
+        TimeSpan? timeOut = null, int requeueCount = -1, TimeSpan? requeueDelay = null, int unacceptableMessageLimit = 0,
+        MessagePumpType messagePumpType = MessagePumpType.Unknown, IAmAChannelFactory? channelFactory = null,
+        OnMissingChannel makeChannels = OnMissingChannel.Create, TimeSpan? emptyChannelDelay = null, 
+        TimeSpan? channelFailureDelay = null,
+        ISchema<ReadOnlySequence<byte>>? schema = null,
+        SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition.Earliest, 
+        int priorityLevel = 0, 
+        bool readCompacted = false, 
+        SubscriptionType subscriptionType = SubscriptionType.Exclusive, 
+        bool allowOutOfOrderDeliver = false,
+        Action<IConsumerBuilder<ReadOnlySequence<byte>>>? configuration = null) 
+        : base(subscriptionName, channelName, routingKey, requestType, getRequestType, bufferSize, noOfPerformers, timeOut, requeueCount, 
+            requeueDelay, unacceptableMessageLimit, messagePumpType, channelFactory, makeChannels, emptyChannelDelay, 
+            channelFailureDelay)
+    {
+        Schema = schema ?? DotPulsar.Schema.ByteSequence;
+        InitialPosition = initialPosition;
+        PriorityLevel = priorityLevel;
+        ReadCompacted = readCompacted;
+        SubscriptionType = subscriptionType;
+        AllowOutOfOrderDeliver = allowOutOfOrderDeliver;
+        Configuration = configuration;
+    }


❌ New issue: Constructor Over-Injection
PulsarSubscription has 23 arguments, max arguments = 5

_Suppress

codescene-delta-analysis · 2025-12-03T09:55:19Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarSubscription.cs

+    public PulsarSubscription(SubscriptionName subscriptionName, ChannelName channelName, RoutingKey routingKey,
+        Func<Message, Type>? getRequestType = null, int bufferSize = 1, int noOfPerformers = 1, TimeSpan? timeOut = null,
+        int requeueCount = -1, TimeSpan? requeueDelay = null, int unacceptableMessageLimit = 0,
+        MessagePumpType messagePumpType = MessagePumpType.Unknown, IAmAChannelFactory? channelFactory = null,
+        OnMissingChannel makeChannels = OnMissingChannel.Create, TimeSpan? emptyChannelDelay = null,
+        TimeSpan? channelFailureDelay = null,
+        ISchema<ReadOnlySequence<byte>>? schema = null,
+        SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition.Earliest,
+        int priorityLevel = 0,
+        bool readCompacted = false,
+        SubscriptionType subscriptionType = SubscriptionType.Exclusive,
+        bool allowOutOfOrderDeliver = false,
+        Action<IConsumerBuilder<ReadOnlySequence<byte>>>? configuration = null)
+        : base(subscriptionName, channelName, routingKey, typeof(T), getRequestType, bufferSize, noOfPerformers, timeOut, requeueCount,
+            requeueDelay, unacceptableMessageLimit, messagePumpType, channelFactory, makeChannels, emptyChannelDelay,
+            channelFailureDelay, schema, initialPosition, priorityLevel, readCompacted, subscriptionType, 
+            allowOutOfOrderDeliver, configuration)
+    {
+
+    }


❌ New issue: Constructor Over-Injection
PulsarSubscription has 22 arguments, max arguments = 5

_Suppress

codescene-delta-analysis · 2025-12-03T09:55:20Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageProducer.cs

+    private MessageMetadata CreateMessageMetadata(Message message, TimeSpan? delay)
+    {
+        var metadata = new MessageMetadata
+        {
+            Key = PartitionKey.IsNullOrEmpty(message.Header.PartitionKey) ? null : message.Header.PartitionKey.Value,
+            EventTimeAsDateTimeOffset = message.Header.TimeStamp,
+            SchemaVersion = publication.SchemaVersion,
+            SequenceId = publication.GenerateSequenceId(message),
+        };
+
+        if (delay.HasValue && delay.Value != TimeSpan.Zero)
+        {
+            metadata.DeliverAtTimeAsDateTimeOffset = time.GetUtcNow() + delay.Value; 
+        }
+
+        metadata[HeaderNames.ContentType] = message.Header.ContentType.ToString();
+        metadata[HeaderNames.CorrelationId] = message.Header.CorrelationId;
+        metadata[HeaderNames.MessageType] = message.Header.MessageType.ToString();
+        metadata[HeaderNames.MessageId] = message.Header.MessageId;
+        metadata[HeaderNames.SpecVersion] = message.Header.SpecVersion;
+        metadata[HeaderNames.Type] = message.Header.Type;
+        metadata[HeaderNames.Time] = message.Header.TimeStamp.ToRfc3339();
+        metadata[HeaderNames.Topic] = message.Header.Topic;
+        metadata[HeaderNames.Source] = message.Header.Source.ToString();
+        metadata[HeaderNames.Baggage] = message.Header.Baggage.ToString();
+
+        if (!RoutingKey.IsNullOrEmpty(message.Header.ReplyTo))
+        {
+            metadata[HeaderNames.ReplyTo] = message.Header.ReplyTo;
+        }
+
+        if (!string.IsNullOrEmpty(message.Header.Subject))
+        {
+            metadata[HeaderNames.Subject] = message.Header.Subject;
+        }
+
+        if (message.Header.DataSchema != null)
+        {
+            metadata[HeaderNames.DataSchema] = message.Header.DataSchema.ToString();
+        }
+
+        if (!TraceParent.IsNullOrEmpty(message.Header.TraceParent))
+        {
+            metadata[HeaderNames.TraceParent] = message.Header.TraceParent;
+        }
+
+        if (!TraceState.IsNullOrEmpty(message.Header.TraceState))
+        {
+            metadata[HeaderNames.TraceState] = message.Header.TraceState;
+        }
+
+        foreach (var pair in  message.Header.Bag)
+        {
+            metadata[pair.Key] = pair.Value.ToString();
+        }
+
+        return metadata;
+    }


❌ New issue: Complex Method
CreateMessageMetadata has a cyclomatic complexity of 10, threshold = 9

_Suppress

codescene-delta-analysis · 2025-12-03T09:55:20Z

src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumer.cs

+    public async Task AcknowledgeAsync(Message message, CancellationToken cancellationToken = default)
+    { 
+        if (!message.Header.Bag.TryGetValue("ReceiptHandle", out var receiptHandle))
+        {
+            return;
+        }
+
+        if (receiptHandle is not MessageId messageId)
+        {
+            return;
+        }
+
+        try
+        {
+            await backgroundPulsarConsumer.Consumer.Acknowledge(messageId, cancellationToken);
+            Log.AcknowledgedMessage(s_logger, message.Id, messageId.ToString());
+        }
+        catch (Exception ex)
+        {
+            Log.ErrorAcknowledgingMessage(s_logger, ex, message.Id, messageId.ToString());
+            throw;
+        }
+    }


❌ New issue: Code Duplication
The module contains 3 functions with similar structure: AcknowledgeAsync,RejectAsync,RequeueAsync

_Suppress

Rafael Lillo added 6 commits July 23, 2025 19:15

Init support to pulsar

159ee9c

Merge branch 'master' into add-support-pulsar

91083e5

# Conflicts: # Brighter.sln

Add comment

892f467

Add Pulsar unit test

65c3d93

Add comments to header name

2581d23

feat: Update csproj

48e1889

lillo42 self-assigned this Jul 23, 2025

lillo42 added 3 - Done feature request v10 .NET Pull requests that update .net code V10.X labels Jul 23, 2025

This comment was marked as outdated.

Sign in to view

lillo42 added 2 - In Progress and removed 3 - Done labels Jul 23, 2025

fix: build

215f017

This comment was marked as outdated.

Sign in to view

iancooper removed the RC2 label Jul 25, 2025

Rafael Lillo added 2 commits July 28, 2025 18:47

Merge branch 'master' into add-support-pulsar

211bd26

feat: Add Pulsar CI

5ebf049

This comment was marked as outdated.

Sign in to view

fix: error to run pulsar

641f783

This comment was marked as outdated.

Sign in to view

lillo42 added 2 commits August 5, 2025 11:44

Merge branch 'master' into add-support-pulsar

ccd37c7

# Conflicts: # .github/workflows/ci.yml # Brighter.sln # Directory.Packages.props

fix: build

309f4c6

This comment was marked as outdated.

Sign in to view

fix: Remove RocketMQ build & build tests

cb86df3

This comment was marked as outdated.

Sign in to view

fix: pulsar tests

1800a5f

lillo42 mentioned this pull request Aug 15, 2025

refactor(pulsar-gateway): restructure csproj and improve code quality #3720

Closed

Rafael Lillo added 2 commits August 15, 2025 17:06

Merge branch 'master' into add-support-pulsar

68fba51

Update Pulsar docs

5ed1cc1

lillo42 added 3 - Done and removed 2 - In Progress labels Aug 15, 2025

lillo42 marked this pull request as ready for review August 15, 2025 16:27

lillo42 requested review from DevJonny, holytshirt, iancooper and preardon as code owners August 15, 2025 16:27

This comment was marked as outdated.

Sign in to view

lillo42 added 2 commits September 16, 2025 14:32

Merge branch 'master' into add-support-pulsar

7cf9466

# Conflicts: # .github/workflows/ci.yml

Change the max of message same as NoOfPerformermers

afe0fa3

This comment was marked as outdated.

Sign in to view

Merge branch 'master' into add-support-pulsar

bbf95bf

This comment was marked as outdated.

Sign in to view

Rafael Lillo and others added 4 commits September 26, 2025 19:00

Merge branch 'master' into add-support-pulsar

afe30ba

Merge branch 'master' into add-support-pulsar

777734e

Merge branch 'master' into add-support-pulsar

00ecbbc

# Conflicts: # .github/workflows/ci.yml

Revert unnecessary changes

f302def

codescene-delta-analysis bot reviewed Oct 3, 2025

View reviewed changes

iancooper reviewed Oct 24, 2025

View reviewed changes

lillo42 added 2 - In Progress Draft This is a work in progress and removed 3 - Done labels Dec 1, 2025

lillo42 added 3 commits December 2, 2025 09:52

Merge branch 'master' into add-support-pulsar

5f44003

# Conflicts: # Brighter.sln # src/Paramore.Brighter/Observability/MessagingSystem.cs

Merge with master

34efccd

Updata pacakges

443f065

codescene-delta-analysis bot reviewed Dec 3, 2025

View reviewed changes

iancooper removed the v10 label Dec 31, 2025


		var consumer = builder.Create();

		return new PulsarBackgroundMessageConsumer(pulsarSubscription.NoOfPerformers, consumer);

feat: Add support Apache Pulsar #3682

Are you sure you want to change the base?

feat: Add support Apache Pulsar #3682

Uh oh!

Conversation

lillo42 commented Jul 23, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

codescene-delta-analysis bot left a comment

Choose a reason for hiding this comment

Uh oh!

codescene-delta-analysis bot Oct 3, 2025

Choose a reason for hiding this comment

Uh oh!

codescene-delta-analysis bot Oct 3, 2025

Choose a reason for hiding this comment

Uh oh!

iancooper left a comment

Choose a reason for hiding this comment

Uh oh!

iancooper Oct 24, 2025

Choose a reason for hiding this comment

Uh oh!

iancooper Oct 24, 2025

Choose a reason for hiding this comment

Uh oh!

iancooper Oct 24, 2025

Choose a reason for hiding this comment

Uh oh!

iancooper Oct 24, 2025

Choose a reason for hiding this comment

Uh oh!

codescene-delta-analysis bot left a comment

Choose a reason for hiding this comment

Uh oh!

codescene-delta-analysis bot Dec 3, 2025

Choose a reason for hiding this comment

Uh oh!

codescene-delta-analysis bot Dec 3, 2025

Choose a reason for hiding this comment

Uh oh!

codescene-delta-analysis bot Dec 3, 2025

Choose a reason for hiding this comment

Uh oh!

codescene-delta-analysis bot Dec 3, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants