Combined: broker tests in CI, ActiveMQ image, MsgPack deprecation, Elasticsearch and NATS.Net ports, Kafka subscribe fix

.github/workflows/dotnet.yml

@@ -17,6 +17,39 @@ jobs:
       contents: read
       id-token: write # NuGet Trusted Publishing
 
+    # Brokers for the transport tests marked Category=Explicit
+    services:
+      rabbitmq:
+        image: rabbitmq:4.4-alpine
+        ports: ['5672:5672']
+        options: >-
+          --health-cmd "rabbitmq-diagnostics -q check_port_connectivity"
+          --health-interval 5s --health-timeout 5s --health-retries 20
+      nats:
+        image: nats:2.15
+        ports: ['4222:4222']
+      kafka:
+        image: apache/kafka:4.4.0
+        ports: ['9092:9092']
+        # Setting any KAFKA_* variable replaces the image's defaults, so configure a full single-node KRaft broker
+        env:
+          KAFKA_NODE_ID: 1
+          KAFKA_PROCESS_ROLES: broker,controller
+          KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
+          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+          KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
+          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+          KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS: 1
+          KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+          KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+          KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1
+          KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 1
+        options: >-
+          --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092"
+          --health-interval 5s --health-timeout 10s --health-retries 20
+
     steps:
     - name: Checkout
       uses: actions/checkout@v7
@@ -40,6 +73,17 @@ jobs:
       run: dotnet restore
     - name: Build
       run: dotnet build --no-restore --configuration Release -p:Version=${{ steps.gitversion.outputs.semVer }} -p:ContinuousIntegrationBuild=true -p:Deterministic=true
+    - name: Create Kafka test topic
+      run: docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic test-topic-3
+    - name: Test transports against brokers
+      env:
+        RABBITMQ_URI: amqp://localhost:5672
+        NATS_URL: nats://localhost:4222
+        KAFKA_BOOTSTRAP_SERVERS: localhost:9092
+      run: |
+        dotnet test --project Obvs.RabbitMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
+        dotnet test --project Obvs.NATS.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
+        dotnet test --project Obvs.Kafka.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
     - name: Test
       # xUnit v3 runs on Microsoft.Testing.Platform (global.json). Exit code 8 means a project ran no tests,
       # as the transport test projects only contain Explicit tests.

Directory.Packages.props

@@ -7,11 +7,11 @@
         <PackageVersion Include="Apache.NMS.ActiveMQ" Version="2.3.0"/>
         <PackageVersion Include="App.Metrics.Abstractions" Version="4.3.0"/>
         <PackageVersion Include="Confluent.Kafka" Version="2.16.0"/>
+        <PackageVersion Include="Elastic.Clients.Elasticsearch" Version="9.5.3"/>
         <PackageVersion Include="MessagePack" Version="3.1.12"/>
         <PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="10.0.13"/>
         <PackageVersion Include="MsgPack.Cli" Version="1.0.1"/>
-        <PackageVersion Include="NATS.Client" Version="1.1.8"/>
-        <PackageVersion Include="NEST" Version="7.17.5"/>
+        <PackageVersion Include="NATS.Client.Core" Version="3.3.1"/>
         <PackageVersion Include="NetJSON" Version="1.4.7"/>
         <PackageVersion Include="NetMQ" Version="4.0.4.5"/>
         <PackageVersion Include="Newtonsoft.Json" Version="13.0.4"/>

Examples/docker-compose.yml

@@ -1,7 +1,10 @@
-version: '3.5'
 services:
   amq:
-    image: rmohr/activemq:latest
+    image: apache/activemq-classic:6.2.0
+    environment:
+      # Require the credentials the examples connect with
+      - ACTIVEMQ_CONNECTION_USER=admin
+      - ACTIVEMQ_CONNECTION_PASSWORD=admin
     ports:
       - "61616:61616"
       - "8161:8161"
@@ -20,4 +23,4 @@ services:
 
 networks:
   obvs:
-    name: obvs
+    name: obvs

Obvs.Kafka.Tests/TestBroker.cs

@@ -0,0 +1,10 @@
+using System;
+
+namespace Obvs.Kafka.Tests
+{
+    internal static class TestBroker
+    {
+        // Set KAFKA_BOOTSTRAP_SERVERS to run the Explicit tests against a broker other than a default local install
+        public static string BootstrapServers => Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS") ?? "localhost:9092";
+    }
+}

Obvs.Kafka.Tests/TestMessageSendAndReceive.cs

@@ -1,5 +1,7 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Linq;
 using System.Reactive.Linq;
 using System.Threading.Tasks;
 using System.Xml.Serialization;
@@ -44,13 +46,8 @@ public override string ToString()
 
         public TestMessageSendAndReceive()
         {
-            // Steps to run local kafka broker:
-            // - Install: Docker for Windows / Docker Desktop
-            // - Clone: https://github.com/wurstmeister/kafka-docker
-            // - Edit: docker-compose-single-broker.yml
-            // - Change: KAFKA_ADVERTISED_HOST_NAME: localhost
-            // - Run: docker-compose -f "docker-compose-single-broker.yml" up -d --build
-            var cfg = new KafkaConfiguration("localhost:9092");
+            // Run a local broker with: docker run -p 9092:9092 apache/kafka
+            var cfg = new KafkaConfiguration(TestBroker.BootstrapServers);
             var sourceConfig = new KafkaSourceConfiguration();
             var producerConfig = new KafkaProducerConfiguration();
             const string topicName = "test-topic-3";
@@ -100,5 +97,27 @@ public async Task TestSendAndReceiveMessage()
 
             sub.Dispose();
         }
+
+        [Fact]
+        [Trait("Category", "Explicit")]
+        public async Task TestMessagePublishedStraightAfterSubscribingIsReceived()
+        {
+            var data = $"straight after subscribing {Guid.NewGuid()}";
+            var messages = new ConcurrentQueue<TestMessage>();
+
+            using (_source.Messages.OfType<TestMessage>().Subscribe(messages.Enqueue))
+            {
+                // No delay: once Subscribe returns, a published message must not be missed
+                await _publisher.PublishAsync(new TestMessage(data));
+
+                var deadline = DateTime.UtcNow.AddSeconds(10);
+                while (!messages.Any(message => message.Data == data) && DateTime.UtcNow < deadline)
+                {
+                    await Task.Delay(100, TestContext.Current.CancellationToken);
+                }
+            }
+
+            Assert.Contains(messages, message => message.Data == data);
+        }
     }
 }

Obvs.Kafka/MessageSource.cs

@@ -24,6 +24,10 @@ public class MessageSource<TMessage> : IMessageSource<TMessage>
 
         private readonly ConsumerConfig _consumerConfig;
 
+        // How long Subscribe waits for Kafka to assign partitions, and for each offset query while it does
+        private static readonly TimeSpan PartitionAssignmentTimeout = TimeSpan.FromSeconds(30);
+        private static readonly TimeSpan OffsetQueryTimeout = TimeSpan.FromSeconds(10);
+
         public MessageSource(KafkaConfiguration kafkaConfig,
             KafkaSourceConfiguration sourceConfig, 
             string topicName,
@@ -60,7 +64,8 @@ public IObservable<TMessage> Messages
 
                     var task = Task.Run(() => Subscriber(token, observer, subscribedEvent));
 
-                    subscribedEvent.Wait();
+                    // Wait until partitions are assigned, so messages published once Subscribe returns are received
+                    subscribedEvent.Wait(PartitionAssignmentTimeout);
 
                     return Disposable.Create(() =>
                     {
@@ -78,10 +83,15 @@ private void Subscriber(CancellationToken token, IObserver<TMessage> observer, M
                 using (var consumer = new ConsumerBuilder<Ignore, WrapperMessage>(_consumerConfig)
                     .SetValueDeserializer(new ConsumerValueDeserializer<WrapperMessage>())
                     .SetErrorHandler((_, e) => observer.OnError(new Exception($"Consumer ErrorHandler: {e.Reason}")))
+                    .SetPartitionsAssignedHandler((c, partitions) =>
+                    {
+                        var offsets = StartOffsets(c, partitions);
+                        subscribedEvent.Set();
+                        return offsets;
+                    })
                     .Build())
                 {
                     consumer.Subscribe(_topicName);
-                    subscribedEvent.Set();
 
                     try
                     {
@@ -114,6 +124,17 @@ private void Subscriber(CancellationToken token, IObserver<TMessage> observer, M
             }
         }
 
+        // Fix each partition's start position at assignment: the group's committed offset, or else the current
+        // end of the log. Leaving "latest" to be resolved later would skip messages published in the meantime.
+        private static IEnumerable<TopicPartitionOffset> StartOffsets(IConsumer<Ignore, WrapperMessage> consumer, List<TopicPartition> partitions)
+        {
+            return consumer.Committed(partitions, OffsetQueryTimeout)
+                .Select(committed => committed.Offset != Offset.Unset
+                    ? committed
+                    : new TopicPartitionOffset(committed.TopicPartition, consumer.QueryWatermarkOffsets(committed.TopicPartition, OffsetQueryTimeout).High))
+                .ToList();
+        }
+
         private void ConsumeMessage(CancellationToken token, IObserver<TMessage> observer, IConsumer<Ignore, WrapperMessage> consumer)
         {
             try

Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs

@@ -1,8 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.Reactive.Concurrency;
-using Elasticsearch.Net;
-using Nest;
+using Elastic.Clients.Elasticsearch;
+using Elastic.Transport;
 using Obvs.Configuration;
 
 namespace Obvs.Monitoring.ElasticSearch.Configuration
@@ -57,7 +57,7 @@ public interface ICanSpecifyElasticSearchMonitoringUri<TMessage, TCommand, TEven
     {
         ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(string uri);
 
-        ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(IConnectionPool elasticConnectionPool);
+        ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(NodePool elasticConnectionPool);
     }
 
     internal class ElasticSearchMonitoringConfig<TMessage, TCommand, TEvent, TRequest, TResponse> :
@@ -94,18 +94,18 @@ public ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent,
             var monitorFactory = new ElasticSearchMonitorFactory<TMessage>(
                 _indexPrefix, _types, _instanceName,
                 _samplePeriod, Scheduler.Default, 
-                new ElasticClient(new ConnectionSettings(new Uri(uri))));
+                new ElasticsearchClient(new ElasticsearchClientSettings(new Uri(uri))));
 
             _config.UsingMonitor(monitorFactory);
             return _config;
         }
 
-        public ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(IConnectionPool elasticConnectionPool)
+        public ICanAddEndpointOrLoggingOrCorrelationOrCreate<TMessage, TCommand, TEvent, TRequest, TResponse> ConnectToServer(NodePool elasticConnectionPool)
         {
             var monitorFactory = new ElasticSearchMonitorFactory<TMessage>(
                 _indexPrefix, _types, _instanceName,
                 _samplePeriod, Scheduler.Default,
-                new ElasticClient(new ConnectionSettings(elasticConnectionPool)));
+                new ElasticsearchClient(new ElasticsearchClientSettings(elasticConnectionPool)));
 
             _config.UsingMonitor(monitorFactory);
             return _config;

Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs

@@ -6,7 +6,7 @@
 using System.Reactive.Concurrency;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
-using Nest;
+using Elastic.Clients.Elasticsearch;
 
 namespace Obvs.Monitoring.ElasticSearch
 {
@@ -22,7 +22,7 @@ static class Directions
         private readonly string _counterName;
         private readonly string _indexPrefix;
         private readonly IList<Type> _types;
-        private readonly IElasticClient _client;
+        private readonly ElasticsearchClient _client;
         private readonly Subject<Tuple<TMessage, TimeSpan>> _queueSent = new Subject<Tuple<TMessage, TimeSpan>>(); 
         private readonly Subject<Tuple<TMessage, TimeSpan>> _queueReceived = new Subject<Tuple<TMessage, TimeSpan>>();
         private readonly TimeSpan _samplePeriod;
@@ -34,7 +34,7 @@ static class Directions
         private readonly string _processName;
         private readonly IList<ObvsCounter> _emptyList = new List<ObvsCounter>();
 
-        public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList<Type> types, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client)
+        public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList<Type> types, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client)
         {
             _messageTypeName = typeof(TMessage).Name;
             _instanceName = instanceName;
@@ -69,9 +69,9 @@ private void Save(IList<ObvsCounter> items)
                 {
                     string indexName = string.Format("{0}-{1}", _indexPrefix, DateTime.Today.ToString("yyyy.MM.dd"));
                     var response = _client.IndexMany(items, indexName);
-                    if (!response.IsValid && response.ServerError != null)
+                    if (!response.IsValidResponse && response.ElasticsearchServerError != null)
                     {
-                        Console.WriteLine(response.ServerError.Error);
+                        Console.WriteLine(response.ElasticsearchServerError.Error);
                     }
                 }
             }

Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Reactive.Concurrency;
-using Nest;
+using Elastic.Clients.Elasticsearch;
 
 namespace Obvs.Monitoring.ElasticSearch
 {
@@ -12,9 +12,9 @@ public class ElasticSearchMonitorFactory<TMessage> : IMonitorFactory<TMessage>
         private readonly string _instanceName;
         private readonly TimeSpan _samplePeriod;
         private readonly IScheduler _scheduler;
-        private readonly IElasticClient _client;
+        private readonly ElasticsearchClient _client;
 
-        public ElasticSearchMonitorFactory(string indexPrefx, IList<Type> types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client)
+        public ElasticSearchMonitorFactory(string indexPrefx, IList<Type> types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client)
         {
             _indexPrefx = indexPrefx;
             _types = types;

Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj

@@ -11,7 +11,7 @@
     <Description>Monitoring extension for Obvs, using Elastic.</Description>
   </PropertyGroup>
   <ItemGroup>
-    <PackageReference Include="NEST" />
+    <PackageReference Include="Elastic.Clients.Elasticsearch" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\Obvs\Obvs.csproj" />

Obvs.Monitoring.ElasticSearch/ObvsCounter.cs

@@ -1,5 +1,4 @@
 using System;
-using Nest;
 
 namespace Obvs.Monitoring.ElasticSearch
 {