Pt/finish kafka (#5)

* Add kafka eventbus to Catalog, Ordering and Payment.

Only catalog is confirmed to work, I think the ordering background
service and payment are still subscribing to rabbitMQ for some reason
(needs more investigation)

* Fix kafka consumer group ids.

* Fix build problems (duplicate package + lower version)

All services seem to use Kafka now, but it seems like
kafka takes to long to start up and the services
fail to connect to the borkers (either we have to do
retries similar like rabbitmq) or we are going to
enforce the execution order in docker compose by
using something like healthchecks.

* Add kafka broker dependency for other services.

Still have bug in eventBus subscription because
for example the ordering service should
handle a UserCheckoutAccepted event put it does have
no subscription for it when such an event is
published to the Kafka topic:

```
src-ordering-api-1              | [15:04:42 WRN] No subscription for Kafka event: UserCheckoutAccepted
src-ordering-api-1              | Consumed event: UserCheckoutAccepted
src-ordering-api-1              |  Content: 632B63DB0CE145D499FE01904F76A475
```

* Add logging for subscription manager problem.

Seems like the subscription manager is not used
correctly in the kafka eventbus (two different objects?).

* add printing handlers

* actually trigger printing handlers

* add kafkapersistentconnection registration

* Revert "add kafkapersistentconnection registration"

This reverts commit 704ee3e36f09f3f3ad48057de31996654a8e3894.

* add allowAutoCreateTopics in consumers (different than default) and in producers (just to be explicit)

* register DefaultKafkaPersistentConnection in ordering.backgroundtasks

* remove noise in logs

* Make eventNames in kafka eventbus consistent.

Do not remove IntegrationEventSuffix, before this
change the subscription handlers and eventNames did not match.

* Create kafka admin background service to create empty topic.

We have to create the eshop_event_bus kafka topic on startup,
because otherwise the consumer of the microservices would fail.

---------

Co-authored-by: Philipp Theyssen <p.theyssen@gmail.com>
Co-authored-by: Philipp Theyssen <34607877+PTheyssen@users.noreply.github.com>

src/BuildingBlocks/EventBus/EventBus/IEventBusSubscriptionsManager.cs

@@ -4,6 +4,7 @@ public interface IEventBusSubscriptionsManager
 {
     bool IsEmpty { get; }
     event EventHandler<string> OnEventRemoved;
+    
     void AddDynamicSubscription<TH>(string eventName)
         where TH : IDynamicIntegrationEventHandler;
 

src/BuildingBlocks/EventBus/EventBus/InMemoryEventBusSubscriptionsManager.cs

@@ -2,8 +2,6 @@
 
 public partial class InMemoryEventBusSubscriptionsManager : IEventBusSubscriptionsManager
 {
-
-
     private readonly Dictionary<string, List<SubscriptionInfo>> _handlers;
     private readonly List<Type> _eventTypes;
 
@@ -30,12 +28,17 @@ public partial class InMemoryEventBusSubscriptionsManager : IEventBusSubscriptio
     {
         var eventName = GetEventKey<T>();
 
+        Console.WriteLine($"Adding Handler: {typeof(TH)} for Event: {eventName}");
         DoAddSubscription(typeof(TH), eventName, isDynamic: false);
 
         if (!_eventTypes.Contains(typeof(T)))
         {
             _eventTypes.Add(typeof(T));
         }
+
+        foreach (var h in _handlers) {
+            Console.WriteLine($"Handler has Subscription for: {h.Key}");
+        }
     }
 
     private void DoAddSubscription(Type handlerType, string eventName, bool isDynamic)

src/BuildingBlocks/EventBus/EventBusKafka/EventBusKafka.cs

@@ -26,7 +26,7 @@ public class EventBusKafka : IEventBus, IDisposable
     
     public void Publish(IntegrationEvent @event)
     {
-        var eventName = @event.GetType().Name.Replace(IntegrationEventSuffix, "");
+        var eventName = @event.GetType().Name;
         var jsonMessage = JsonSerializer.Serialize(@event, @event.GetType());
         
         // map Integration event to kafka message
@@ -47,7 +47,12 @@ public class EventBusKafka : IEventBus, IDisposable
 
         _logger.LogInformation("Subscribing to event {EventName} with {EventHandler}", eventName, typeof(TH).GetGenericTypeName());
 
-        _subsManager.AddSubscription<T, TH>();
+        try {
+            _subsManager.AddSubscription<T, TH>();
+        } catch (Exception e)
+        {
+            Console.WriteLine($"Failed to add subscription {eventName}, because: {e.Message}");
+        }
     }
 
     public void SubscribeDynamic<TH>(string eventName) where TH : IDynamicIntegrationEventHandler

src/BuildingBlocks/EventBus/EventBusKafka/KafkaAdminBackgroundService.cs

@@ -0,0 +1,43 @@
+using Confluent.Kafka.Admin;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Hosting;
+
+namespace Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
+
+public class KafkaAdminBackgroundService : BackgroundService
+{
+    private const string TopicName = "eshop_event_bus";
+    private readonly ILogger<KafkaAdminBackgroundService> _logger;
+    private readonly IConfiguration _configuration;
+
+    public KafkaAdminBackgroundService(
+        IConfiguration configuration,
+        ILogger<KafkaAdminBackgroundService> logger)
+    {
+        _configuration = configuration;
+        _logger = logger;
+    }
+
+    protected override Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        return Task.Run(CreateTopics, stoppingToken);
+    }
+
+    private async Task CreateTopics()
+    {
+        var adminConfig = new AdminClientConfig();
+        _configuration.GetSection("Kafka:AdminSettings").Bind(adminConfig);
+        using (var adminClient = new AdminClientBuilder(adminConfig).Build())
+        {
+            try
+            {
+                await adminClient.CreateTopicsAsync(new TopicSpecification[] {
+                    new TopicSpecification { Name = TopicName, ReplicationFactor = 1, NumPartitions = 1 } });
+            }
+            catch (CreateTopicsException e)
+            {
+                Console.WriteLine($"An error occured creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
+            }
+        }
+    }
+}

src/BuildingBlocks/EventBus/EventBusKafka/KafkaConsumerBackgroundService.cs

@@ -1,4 +1,3 @@
-using System.Security.Cryptography;
 using Autofac;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Hosting;
@@ -47,8 +46,6 @@ public class KafkaConsumerBackgroundService : BackgroundService
                 var eventName = consumeResult.Message.Key;
                 var messageContent = consumeResult.Message.Value;
 
-                Console.WriteLine($"Consumed event: {eventName}\n Content: {Utils.CalculateMd5Hash(messageContent)}");
-
                 if (!_subsManager.HasSubscriptionsForEvent(eventName))
                 {
                     _logger.LogWarning("No subscription for Kafka event: {EventName}", eventName);

src/Services/Basket/Basket.API/Basket.API.csproj

@@ -16,7 +16,6 @@
   <ItemGroup>   
     <PackageReference Include="Microsoft.AspNetCore.Authentication.OpenIdConnect" Version="7.0.0" />   
     <PackageReference Include="System.IdentityModel.Tokens.Jwt" Version="6.25.0" />
-	  <PackageReference Include="Azure.Identity" Version="1.5.0-beta.3" />
     <PackageReference Include="AspNetCore.HealthChecks.AzureServiceBus" Version="6.0.4" />
     <PackageReference Include="AspNetCore.HealthChecks.Rabbitmq" Version="6.0.2" />
     <PackageReference Include="AspNetCore.HealthChecks.Redis" Version="6.0.4" />

src/Services/Basket/Basket.API/Dockerfile.develop

@@ -8,7 +8,7 @@ WORKDIR /src
 
 COPY ["BuildingBlocks/EventBus/EventBus/EventBus.csproj", "BuildingBlocks/EventBus/EventBus/"]
 COPY ["BuildingBlocks/EventBus/EventBusRabbitMQ/EventBusRabbitMQ.csproj", "BuildingBlocks/EventBus/EventBusRabbitMQ/"]
-COPY ["BuildingBlocks/EventBus/EventBusKafka/EventBusKafka.csproj" "BuildingBlocks/EventBus/EvenBusKafka/"]
+COPY ["BuildingBlocks/EventBus/EventBusKafka/EventBusKafka.csproj" "BuildingBlocks/EventBus/EventBusKafka/"]
 COPY ["BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.csproj", "BuildingBlocks/EventBus/EventBusServiceBus/"]
 COPY ["Services/Basket/Basket.API/Basket.API.csproj", "Services/Basket/Basket.API/"]
 COPY ["NuGet.config", "NuGet.config"]

src/Services/Basket/Basket.API/Startup.cs

@@ -262,6 +262,7 @@ public class Startup
         }
         else if (Configuration.GetValue<bool>("KafkaEnabled"))
         {
+            services.AddHostedService<KafkaAdminBackgroundService>();
             services.AddHostedService<KafkaConsumerBackgroundService>();
             services.AddSingleton<IEventBus, EventBusKafka>();
         }

src/Services/Basket/Basket.API/appsettings.Development.json

@@ -16,12 +16,17 @@
   "EventBusConnection": "localhost",
   "KafkaEnabled": true,
   "Kafka": {
-    "ProducerSettings": {
+    "AdminSettings": {
       "BootstrapServers": "broker:9092"
     },
+    "ProducerSettings": {
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
+    },
     "ConsumerSettings": {
       "BootstrapServers": "broker:9092",
-      "GroupId": "basket-group-id"
+      "GroupId": "basket-group-id",
+      "AllowAutoCreateTopics": true
     }
   }
 }

src/Services/Basket/Basket.API/appsettings.json

@@ -29,12 +29,17 @@
   },
   "KafkaEnabled": true,
   "Kafka": {
-    "ProducerSettings": {
+    "AdminSettings": {
       "BootstrapServers": "broker:9092"
     },
+    "ProducerSettings": {
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
+    },
     "ConsumerSettings": {
       "BootstrapServers": "broker:9092",
-      "GroupId": "basket-group-id"
+      "GroupId": "basket-group-id",
+      "AllowAutoCreateTopics": true
     }
   }
 }

src/Services/Catalog/Catalog.API/Catalog.API.csproj

@@ -78,6 +78,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusKafka\EventBusKafka.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusRabbitMQ\EventBusRabbitMQ.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusServiceBus\EventBusServiceBus.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBus\EventBus.csproj" />

src/Services/Catalog/Catalog.API/Dockerfile.develop

@@ -8,6 +8,7 @@ WORKDIR /src
 
 COPY ["BuildingBlocks/EventBus/EventBus/EventBus.csproj", "BuildingBlocks/EventBus/EventBus/"]
 COPY ["BuildingBlocks/EventBus/EventBusRabbitMQ/EventBusRabbitMQ.csproj", "BuildingBlocks/EventBus/EventBusRabbitMQ/"]
+COPY ["BuildingBlocks/EventBus/EventBusKafka/EventBusKafka.csproj" "BuildingBlocks/EventBus/EventBusKafka/"]
 COPY ["BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.csproj", "BuildingBlocks/EventBus/EventBusServiceBus/"]
 COPY ["BuildingBlocks/EventBus/IntegrationEventLogEF/IntegrationEventLogEF.csproj", "BuildingBlocks/EventBus/IntegrationEventLogEF/"]
 COPY ["BuildingBlocks/WebHostCustomization/WebHost.Customization/WebHost.Customization.csproj", "BuildingBlocks/WebHostCustomization/WebHost.Customization/"]

src/Services/Catalog/Catalog.API/GlobalUsings.cs

@@ -19,6 +19,7 @@ global using Microsoft.EntityFrameworkCore.Design;
 global using Microsoft.EntityFrameworkCore.Metadata.Builders;
 global using Microsoft.EntityFrameworkCore;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
+global using Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
 global using Microsoft.eShopOnContainers.BuildingBlocks.IntegrationEventLogEF.Services;
 global using Microsoft.eShopOnContainers.BuildingBlocks.IntegrationEventLogEF.Utilities;

src/Services/Catalog/Catalog.API/Startup.cs

@@ -1,3 +1,5 @@
+using Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
+
 namespace Microsoft.eShopOnContainers.Services.Catalog.API;
 
 public class Startup
@@ -154,6 +156,10 @@ public static class CustomExtensionMethods
                     name: "catalog-servicebus-check",
                     tags: new string[] { "servicebus" });
         }
+        else if (configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            // TODO: might want to add health check
+        }
         else
         {
             hcBuilder
@@ -251,6 +257,10 @@ public static class CustomExtensionMethods
                 return new DefaultServiceBusPersisterConnection(serviceBusConnection);
             });
         }
+        else if (configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            services.AddSingleton<IKafkaPersistentConnection, DefaultKafkaPersistentConnection>();
+        }
         else
         {
             services.AddSingleton<IRabbitMQPersistentConnection>(sp =>
@@ -304,6 +314,11 @@ public static class CustomExtensionMethods
             });
 
         }
+        else if (configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            services.AddHostedService<KafkaConsumerBackgroundService>();
+            services.AddSingleton<IEventBus, EventBusKafka>();
+        }
         else
         {
             services.AddSingleton<IEventBus, EventBusRabbitMQ>(sp =>

src/Services/Catalog/Catalog.API/appsettings.Development.json

@@ -12,13 +12,16 @@
     }
   },
   "EventBusConnection": "localhost",
+  "KafkaEnabled": true,
   "Kafka": {
     "ProducerSettings": {
-      "BootstrapServers": "localhost:9092"
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
     },
     "ConsumerSettings": {
-      "BootstrapServers": "localhost:9092",
+      "BootstrapServers": "broker:9092",
-      "GroupId": "catalog-group-id"
+      "GroupId": "catalog-group-id",
+      "AllowAutoCreateTopics": true
     }
   }
 }

src/Services/Catalog/Catalog.API/appsettings.json

@@ -24,7 +24,18 @@
     "Name": "eshop",
     "ClientId": "your-client-id",
     "ClientSecret": "your-client-secret"
+  },
+  "KafkaEnabled": true,
+  "Kafka": {
+    "ProducerSettings": {
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
+    },
+    "ConsumerSettings": {
+      "BootstrapServers": "broker:9092",
+      "GroupId": "catalog-group-id",
+      "AllowAutoCreateTopics": true
+    }
   }
-
 }
   

src/Services/Ordering/Ordering.API/Dockerfile.develop

@@ -6,6 +6,7 @@ EXPOSE 80
 
 WORKDIR /src
 COPY ["BuildingBlocks/EventBus/EventBus/EventBus.csproj", "BuildingBlocks/EventBus/EventBus/"]
+COPY "BuildingBlocks/EventBus/EventBusKafka/EventBusKafka.csproj" "BuildingBlocks/EventBus/EventBusKafka/EventBusKafka.csproj"
 COPY ["BuildingBlocks/EventBus/EventBusRabbitMQ/EventBusRabbitMQ.csproj", "BuildingBlocks/EventBus/EventBusRabbitMQ/"]
 COPY ["BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.csproj", "BuildingBlocks/EventBus/EventBusServiceBus/"]
 COPY ["BuildingBlocks/EventBus/IntegrationEventLogEF/IntegrationEventLogEF.csproj", "BuildingBlocks/EventBus/IntegrationEventLogEF/"]

src/Services/Ordering/Ordering.API/GlobalUsings.cs

@@ -27,6 +27,7 @@ global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Extensions;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
+global using Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
 global using Microsoft.eShopOnContainers.BuildingBlocks.IntegrationEventLogEF.Services;

src/Services/Ordering/Ordering.API/Ordering.API.csproj

@@ -26,6 +26,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusKafka\EventBusKafka.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusRabbitMQ\EventBusRabbitMQ.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusServiceBus\EventBusServiceBus.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBus\EventBus.csproj" />

src/Services/Ordering/Ordering.API/Startup.cs

@@ -1,5 +1,6 @@
 using Microsoft.AspNetCore.Authentication.Cookies;
 using Microsoft.AspNetCore.Authentication.OpenIdConnect;
+using Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
 
 namespace Microsoft.eShopOnContainers.Services.Ordering.API;
 
@@ -174,6 +175,10 @@ static class CustomExtensionsMethods
                     name: "ordering-servicebus-check",
                     tags: new string[] { "servicebus" });
         }
+        else if (configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            // TODO: might want to add health check
+        }
         else
         {
             hcBuilder
@@ -270,6 +275,10 @@ static class CustomExtensionsMethods
                 return new DefaultServiceBusPersisterConnection(serviceBusConnectionString);
             });
         }
+        else if (configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            services.AddSingleton<IKafkaPersistentConnection, DefaultKafkaPersistentConnection>();
+        }
         else
         {
             services.AddSingleton<IRabbitMQPersistentConnection>(sp =>
@@ -347,6 +356,11 @@ static class CustomExtensionsMethods
                     eventBusSubcriptionsManager, iLifetimeScope, subscriptionName);
             });
         }
+        else if (configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            services.AddHostedService<KafkaConsumerBackgroundService>();
+            services.AddSingleton<IEventBus, EventBusKafka>();
+        }
         else
         {
             services.AddSingleton<IEventBus, EventBusRabbitMQ>(sp =>

src/Services/Ordering/Ordering.API/appsettings.json

@@ -23,6 +23,18 @@
   },
   "EventBusRetryCount": 5,
   "EventBusConnection": "localhost",
+  "KafkaEnabled": true,
+  "Kafka": {
+    "ProducerSettings": {
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
+    },
+    "ConsumerSettings": {
+      "BootstrapServers": "broker:9092",
+      "GroupId": "ordering-group-id",
+      "AllowAutoCreateTopics": true
+    }
+  },
   "UseVault": false,
   "Vault": {
     "Name": "eshop",

src/Services/Ordering/Ordering.BackgroundTasks/Extensions/CustomExtensionMethods.cs

@@ -2,6 +2,7 @@
 using Azure.Messaging.ServiceBus;
 using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
 using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
+using Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
 using Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ;
 using Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
 using Microsoft.Extensions.Configuration;
@@ -34,6 +35,10 @@ namespace Ordering.BackgroundTasks.Extensions
                         name: "orderingtask-servicebus-check",
                         tags: new string[] { "servicebus" });
             }
+            else if (configuration.GetValue<bool>("KafkaEnabled"))
+            {
+                // TODO: might want to add health check
+            }
             else
             {
                 hcBuilder.AddRabbitMQ(
@@ -69,6 +74,12 @@ namespace Ordering.BackgroundTasks.Extensions
                     return new EventBusServiceBus(serviceBusPersisterConnection, logger, eventBusSubcriptionsManager, iLifetimeScope, subscriptionName);
                 });
             }
+            else if (configuration.GetValue<bool>("KafkaEnabled"))
+            {
+                services.AddSingleton<IKafkaPersistentConnection, DefaultKafkaPersistentConnection>();
+                services.AddHostedService<KafkaConsumerBackgroundService>();
+                services.AddSingleton<IEventBus, EventBusKafka>();
+            }
             else
             {
                 services.AddSingleton<IRabbitMQPersistentConnection>(sp =>

src/Services/Ordering/Ordering.BackgroundTasks/Ordering.BackgroundTasks.csproj

@@ -16,7 +16,7 @@
     <PackageReference Include="Autofac" Version="6.5.0" />
     <PackageReference Include="Dapper" Version="2.0.123" />
     <PackageReference Include="Microsoft.AspNetCore.Diagnostics.HealthChecks" Version="2.2.0" />
-    <PackageReference Include="Microsoft.Extensions.Hosting" Version="7.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="7.0.1" />
     <PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.17.0" />
     <PackageReference Include="Serilog.AspNetCore" Version="6.1.0-dev-00289" />
     <PackageReference Include="Serilog.Enrichers.Environment" Version="2.2.1-dev-00787" />
@@ -28,6 +28,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusKafka\EventBusKafka.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusRabbitMQ\EventBusRabbitMQ.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusServiceBus\EventBusServiceBus.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBus\EventBus.csproj" />

src/Services/Ordering/Ordering.BackgroundTasks/appsettings.Development.json

@@ -6,13 +6,16 @@
       "Microsoft": "Information"
     }
   },
+  "KafkaEnabled": true,
   "Kafka": {
     "ProducerSettings": {
-      "BootstrapServers": "localhost:9092"
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
     },
     "ConsumerSettings": {
-      "BootstrapServers": "localhost:9092",
+      "BootstrapServers": "broker:9092",
-      "GroupId": "ordering-group-id"
+      "GroupId": "ordering-background-group-id",
+      "AllowAutoCreateTopics": true
     }
   }
 }

src/Services/Ordering/Ordering.BackgroundTasks/appsettings.json

@@ -22,5 +22,17 @@
   "EventBusRetryCount": 5,
   "EventBusConnection": "",
   "EventBusUserName": "",
-  "EventBusPassword": ""
+  "EventBusPassword": "",
+  "KafkaEnabled": true,
+  "Kafka": {
+    "ProducerSettings": {
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
+    },
+    "ConsumerSettings": {
+      "BootstrapServers": "broker:9092",
+      "GroupId": "ordering-background-group-id",
+      "AllowAutoCreateTopics": true
+    }
+  }
 }

src/Services/Payment/Payment.API/Dockerfile.develop

@@ -7,6 +7,7 @@ EXPOSE 80
 WORKDIR /src
 COPY ["BuildingBlocks/EventBus/EventBus/EventBus.csproj", "BuildingBlocks/EventBus/EventBus/"]
 COPY ["BuildingBlocks/EventBus/EventBusRabbitMQ/EventBusRabbitMQ.csproj", "BuildingBlocks/EventBus/EventBusRabbitMQ/"]
+COPY ["BuildingBlocks/EventBus/EventBusKafka/EventBusKafka.csproj" "BuildingBlocks/EventBus/EventBusKafka/"]
 COPY ["BuildingBlocks/EventBus/EventBusServiceBus/EventBusServiceBus.csproj", "BuildingBlocks/EventBus/EventBusServiceBus/"]
 COPY ["BuildingBlocks/EventBus/IntegrationEventLogEF/IntegrationEventLogEF.csproj", "BuildingBlocks/EventBus/IntegrationEventLogEF/"]
 COPY ["Services/Payment/Payment.API/Payment.API.csproj", "Services/Payment/Payment.API/"]

src/Services/Payment/Payment.API/GlobalUsings.cs

@@ -10,6 +10,7 @@ global using Azure.Messaging.ServiceBus;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
+global using Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ;
 global using Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
 global using Microsoft.eShopOnContainers.Payment.API.IntegrationEvents.Events;

src/Services/Payment/Payment.API/Payment.API.csproj

@@ -29,6 +29,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusKafka\EventBusKafka.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusRabbitMQ\EventBusRabbitMQ.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusServiceBus\EventBusServiceBus.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBus\EventBus.csproj" />

src/Services/Payment/Payment.API/Startup.cs

@@ -1,3 +1,5 @@
+using Microsoft.eShopOnContainers.BuildingBlocks.EventBusKafka;
+
 namespace Microsoft.eShopOnContainers.Payment.API;
 
 public class Startup
@@ -27,6 +29,10 @@ public class Startup
                 return new DefaultServiceBusPersisterConnection(serviceBusConnectionString);
             });
         }
+        else if (Configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            services.AddSingleton<IKafkaPersistentConnection, DefaultKafkaPersistentConnection>();
+        }
         else
         {
             services.AddSingleton<IRabbitMQPersistentConnection>(sp =>
@@ -116,6 +122,11 @@ public class Startup
                     eventBusSubcriptionsManager, iLifetimeScope, subscriptionName);
             });
         }
+        else if (Configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            services.AddHostedService<KafkaConsumerBackgroundService>();
+            services.AddSingleton<IEventBus, EventBusKafka>();
+        }
         else
         {
             services.AddSingleton<IEventBus, EventBusRabbitMQ>(sp =>
@@ -164,6 +175,10 @@ public static class CustomExtensionMethods
                     name: "payment-servicebus-check",
                     tags: new string[] { "servicebus" });
         }
+        else if (configuration.GetValue<bool>("KafkaEnabled"))
+        {
+            // TODO: might want to add healthcheck topic
+        }
         else
         {
             hcBuilder

src/Services/Payment/Payment.API/appsettings.Development.json

@@ -7,13 +7,16 @@
       "Microsoft": "Information"
     }
   },
+  "KafkaEnabled": true,
   "Kafka": {
     "ProducerSettings": {
-      "BootstrapServers": "localhost:9092"
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
     },
     "ConsumerSettings": {
-      "BootstrapServers": "localhost:9092",
+      "BootstrapServers": "broker:9092",
-      "GroupId": "payment-group-id"
+      "GroupId": "payment-group-id",
+      "AllowAutoCreateTopics": true
     }
   }
 }

src/Services/Payment/Payment.API/appsettings.json

@@ -17,5 +17,17 @@
   "ApplicationInsights": {
     "InstrumentationKey": ""
   },
-  "EventBusRetryCount": 5
+  "EventBusRetryCount": 5,
+  "KafkaEnabled": true,
+  "Kafka": {
+    "ProducerSettings": {
+      "BootstrapServers": "broker:9092",
+      "AllowAutoCreateTopics": true
+    },
+    "ConsumerSettings": {
+      "BootstrapServers": "broker:9092",
+      "GroupId": "payment-group-id",
+      "AllowAutoCreateTopics": true
+    }
+  }
 }

src/docker-compose.yml

@@ -52,6 +52,7 @@ services:
     depends_on:
       - sqldata
       - rabbitmq
+      - broker
 
   ordering-api:
     image: ${REGISTRY:-eshop}/ordering.api:${PLATFORM:-linux}-${TAG:-latest}
@@ -61,6 +62,7 @@ services:
     depends_on:
       - sqldata
       - rabbitmq
+      - broker
 
   ordering-backgroundtasks:
     image: ${REGISTRY:-eshop}/ordering.backgroundtasks:${PLATFORM:-linux}-${TAG:-latest}
@@ -70,6 +72,7 @@ services:
     depends_on:
       - sqldata
       - rabbitmq
+      - broker
 
   payment-api:
     image: ${REGISTRY:-eshop}/payment.api:${PLATFORM:-linux}-${TAG:-latest}
@@ -78,6 +81,7 @@ services:
       dockerfile: Services/Payment/Payment.API/Dockerfile
     depends_on:
       - rabbitmq
+      - broker
 
 #  webhooks-api:
 #    image: ${REGISTRY:-eshop}/webhooks.api:${PLATFORM:-linux}-${TAG:-latest}