feat: Add support Apache Pulsar #3682
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| services: | ||
| # Start zookeeper | ||
| zookeeper: | ||
| image: apachepulsar/pulsar:latest | ||
| container_name: zookeeper | ||
| restart: on-failure | ||
| environment: | ||
| - metadataStoreUrl=zk:zookeeper:2181 | ||
| - PULSAR_MEM=-Xms256m -Xmx256m -XX:MaxDirectMemorySize=256m | ||
| command: | ||
| - bash | ||
| - -c | ||
| - | | ||
| bin/apply-config-from-env.py conf/zookeeper.conf && \ | ||
| bin/generate-zookeeper-config.sh conf/zookeeper.conf && \ | ||
| exec bin/pulsar zookeeper | ||
| healthcheck: | ||
| test: ["CMD", "bin/pulsar-zookeeper-ruok.sh"] | ||
| interval: 10s | ||
| timeout: 5s | ||
| retries: 30 | ||
|
|
||
| # Init cluster metadata | ||
| pulsar-init: | ||
| container_name: pulsar-init | ||
| hostname: pulsar-init | ||
| image: apachepulsar/pulsar:latest | ||
| command: | ||
| - bash | ||
| - -c | ||
| - | | ||
| bin/pulsar initialize-cluster-metadata \ | ||
| --cluster cluster-a \ | ||
| --zookeeper zookeeper:2181 \ | ||
| --configuration-store zookeeper:2181 \ | ||
| --web-service-url http://broker:8080 \ | ||
| --broker-service-url pulsar://broker:6650 | ||
| depends_on: | ||
| zookeeper: | ||
| condition: service_healthy | ||
|
|
||
| # Start bookie | ||
| bookie: | ||
| image: apachepulsar/pulsar:latest | ||
| container_name: bookie | ||
| restart: on-failure | ||
| environment: | ||
| - clusterName=cluster-a | ||
| - zkServers=zookeeper:2181 | ||
| - metadataServiceUri=metadata-store:zk:zookeeper:2181 | ||
| # otherwise every time we run docker compose uo or down we fail to start due to Cookie | ||
| # See: https://github.com/apache/bookkeeper/blob/405e72acf42bb1104296447ea8840d805094c787/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java#L57-68 | ||
| - advertisedAddress=bookie | ||
| - BOOKIE_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m | ||
| depends_on: | ||
| zookeeper: | ||
| condition: service_healthy | ||
| pulsar-init: | ||
| condition: service_completed_successfully | ||
| # Map the local directory to the container to avoid bookie startup failure due to insufficient container disks. | ||
| command: bash -c "bin/apply-config-from-env.py conf/bookkeeper.conf && exec bin/pulsar bookie" | ||
|
|
||
| # Start broker | ||
| broker: | ||
| image: apachepulsar/pulsar:latest | ||
| container_name: broker | ||
| hostname: broker | ||
| restart: on-failure | ||
| environment: | ||
| - metadataStoreUrl=zk:zookeeper:2181 | ||
| - zookeeperServers=zookeeper:2181 | ||
| - clusterName=cluster-a | ||
| - managedLedgerDefaultEnsembleSize=1 | ||
| - managedLedgerDefaultWriteQuorum=1 | ||
| - managedLedgerDefaultAckQuorum=1 | ||
| - advertisedAddress=broker | ||
| - advertisedListeners=external:pulsar://127.0.0.1:6650 | ||
| - PULSAR_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m | ||
| depends_on: | ||
| zookeeper: | ||
| condition: service_healthy | ||
| bookie: | ||
| condition: service_started | ||
| ports: | ||
| - "6650:6650" | ||
| - "8080:8080" | ||
| command: bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker" |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| namespace Paramore.Brighter.MessagingGateway.Pulsar; | ||
|
|
||
| /// <summary> | ||
| /// Contains constant definitions for header names used in messaging systems, | ||
| /// particularly for Apache Pulsar integration with Brighter. | ||
| /// Includes standard headers, CloudEvents (CE) headers, and custom Brighter-Pulsar headers. | ||
| /// </summary> | ||
| public static class HeaderNames | ||
| { | ||
| /// <summary>Content type of the message payload (e.g., application/json)</summary> | ||
| public const string ContentType = "ContentType"; | ||
|
|
||
| /// <summary>Correlation ID for tracing related messages</summary> | ||
| public const string CorrelationId = "CorrelationId"; | ||
|
|
||
| /// <summary>CloudEvents-formatted unique message identifier</summary> | ||
| public const string MessageId = "CE-EventId"; | ||
|
|
||
| /// <summary>Number of times a message has been processed/requeued</summary> | ||
| public const string HandledCount = "HandledCount"; | ||
|
|
||
| /// <summary>Brighter message type classification (e.g., MT_COMMAND, MT_EVENT)</summary> | ||
| public const string MessageType = "MessageType"; | ||
|
|
||
| /// <summary>Reply destination for request-reply patterns</summary> | ||
| public const string ReplyTo = "ReplyTo"; | ||
|
|
||
| /// <summary>CloudEvents specification version (e.g., "1.0")</summary> | ||
| public const string SpecVersion = "CE-SpecVersion"; | ||
|
|
||
| /// <summary>CloudEvents event type descriptor</summary> | ||
| public const string Type = "CE-EventType"; | ||
|
|
||
| /// <summary>Timestamp of event occurrence in RFC3339 format</summary> | ||
| public const string Time = "CE-EventTime"; | ||
|
|
||
| /// <summary>CloudEvents subject describing event content</summary> | ||
| public const string Subject = "CE-Subject"; | ||
|
|
||
| /// <summary>CloudEvents schema URL for payload validation</summary> | ||
| public const string DataSchema = "CE-DataSchema"; | ||
|
|
||
| /// <summary>CloudEvents source URI identifying event origin</summary> | ||
| public const string Source = "CE-Source"; | ||
|
|
||
| /// <summary>W3C Trace Context traceparent value</summary> | ||
| public const string TraceParent = "CE-X-TraceParent"; | ||
|
|
||
| /// <summary>W3C Trace Context tracestate value</summary> | ||
| public const string TraceState = "CE-X-TraceState"; | ||
|
|
||
| /// <summary>OpenTelemetry baggage items (key-value pairs)</summary> | ||
| public const string Baggage = "CE-X-Baggage"; | ||
|
|
||
| /// <summary>Original message topic/routing key</summary> | ||
| public const string Topic = "Topic"; | ||
|
|
||
| /// <summary>Pulsar schema version identifier</summary> | ||
| public const string SchemaVersion = "Brighter-Pulsar-SchemaVersion"; | ||
|
|
||
| /// <summary>Pulsar message sequence identifier</summary> | ||
| public const string SequenceId = "Brighter-Pulsar-SequenceId"; | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFrameworks>$(BrighterTargetFrameworks)</TargetFrameworks> | ||
| <Description>Provides an implementation of the messaging gateway for decoupled invocation in the Paramore.Brighter pipeline, using Apache Pulsar</Description> | ||
| <PackageTags>Apache Pulsar;Pub/Sub;Command;Event;Service Activator;Decoupled;Invocation;Messaging;Remote;Command Dispatcher;Command Processor;Request;Service;Task Queue;Work Queue;Retry;Circuit Breaker;Availability</PackageTags> | ||
| <Authors>Rafael Andrade</Authors> | ||
| <Nullable>enable</Nullable> | ||
| <SignAssembly>false</SignAssembly> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\Paramore.Brighter\Paramore.Brighter.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="DotPulsar" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'"> | ||
| <PackageReference Include="System.Threading.Channels" /> | ||
| </ItemGroup> | ||
| </Project> |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| using System.Buffers; | ||
| using System.Threading; | ||
| using System.Threading.Channels; | ||
| using System.Threading.Tasks; | ||
| using DotPulsar.Abstractions; | ||
|
|
||
| namespace Paramore.Brighter.MessagingGateway.Pulsar; | ||
|
|
||
| /// <summary> | ||
| /// Background message consumer for Apache Pulsar that buffers messages in a bounded channel. | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// This class manages a background message consumption loop that: | ||
| /// <list type="bullet"> | ||
| /// <item><description>Receives messages from Pulsar using an <see cref="IConsumer{T}"/></description></item> | ||
| /// <item><description>Writes messages to a bounded channel for consumption by other components</description></item> | ||
| /// <item><description>Implements reference counting for safe start/stop operations</description></item> | ||
| /// </list> | ||
| /// | ||
| /// The consumer uses a fire-and-forget pattern where the consumption loop runs independently once started. | ||
| /// </remarks> | ||
| /// <param name="maxLenght">Maximum number of messages to buffer in the channel</param> | ||
| /// <param name="consumer">Pulsar message consumer implementation</param> | ||
| public sealed class PulsarBackgroundMessageConsumer(int maxLenght, IConsumer<ReadOnlySequence<byte>> consumer) | ||
| { | ||
| private int _total; | ||
| private CancellationTokenSource? _cancellationTokenSource; | ||
| private readonly Channel<IMessage<ReadOnlySequence<byte>>> _channel = System.Threading.Channels.Channel.CreateBounded<IMessage<ReadOnlySequence<byte>>>(new BoundedChannelOptions(maxLenght) | ||
| { | ||
| SingleReader = false, SingleWriter = true | ||
| }); | ||
|
|
||
| /// <summary> | ||
| /// Provides read access to the message channel | ||
| /// </summary> | ||
| public ChannelReader<IMessage<ReadOnlySequence<byte>>> Reader => _channel.Reader; | ||
|
|
||
| /// <summary> | ||
| /// Gets the underlying Pulsar consumer instance | ||
| /// </summary> | ||
| public IConsumer<ReadOnlySequence<byte>> Consumer => consumer; | ||
|
|
||
| /// <summary> | ||
| /// Starts the background message consumption loop | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// Implements reference counting: | ||
| /// <list type="bullet"> | ||
| /// <item><description>First call starts the background loop</description></item> | ||
| /// <item><description>Subsequent calls increment the reference count but don't start additional loops</description></item> | ||
| /// </list> | ||
| /// </remarks> | ||
| public void Start() | ||
| { | ||
| var total = Interlocked.Increment(ref _total); | ||
| if (total == 1) | ||
| { | ||
| _cancellationTokenSource = new CancellationTokenSource(); | ||
| _ = ExecuteAsync(_cancellationTokenSource.Token); | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Background message consumption loop | ||
| /// </summary> | ||
| /// <param name="cancellationToken">Cancellation token to stop the loop</param> | ||
| /// <remarks> | ||
| /// Continuously performs: | ||
| /// <list type="number"> | ||
| /// <item><description>Receive message from Pulsar</description></item> | ||
| /// <item><description>Write message to output channel</description></item> | ||
| /// <item><description>Wait for channel write availability</description></item> | ||
| /// </list> | ||
| /// | ||
| /// <para>Errors during message reception are silently ignored to maintain loop continuity.</para> | ||
| /// </remarks> | ||
| private async Task ExecuteAsync(CancellationToken cancellationToken) | ||
| { | ||
| while (!cancellationToken.IsCancellationRequested) | ||
| { | ||
| try | ||
| { | ||
| var message = await consumer.Receive(cancellationToken); | ||
|
Member
There was a problem hiding this comment. This feels like a pump within our pump? |
||
| if (message is null) | ||
| { | ||
| continue; | ||
| } | ||
|
|
||
| await _channel.Writer.WriteAsync(message, cancellationToken); | ||
| await _channel.Writer.WaitToWriteAsync(cancellationToken); | ||
| } | ||
| catch | ||
| { | ||
| // Ignoring any errors | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Stops the background message consumption loop | ||
| /// </summary> | ||
| /// <remarks> | ||
| /// Implements reference counting: | ||
| /// <list type="bullet"> | ||
| /// <item><description>Decrements the reference count</description></item> | ||
| /// <item><description>Stops the loop when reference count reaches zero</description></item> | ||
| /// </list> | ||
| /// | ||
| /// Safe to call multiple times - only the last call that brings the count to zero will stop the loop. | ||
| /// </remarks> | ||
| public void Stop() | ||
| { | ||
| var total = Interlocked.Decrement(ref _total); | ||
| if (total == 0 && _cancellationTokenSource != null) | ||
| { | ||
| _cancellationTokenSource.Cancel(); | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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?