Implement Apache Kafka transport for PubSub (Part 14 Annex B.2) with HA (#3942)

Applications/ConsoleReferencePubSubClient/ConsoleReferencePubSubClient.csproj

@@ -22,6 +22,7 @@
     <ProjectReference Include="..\..\Libraries\Opc.Ua.PubSub.Udp\Opc.Ua.PubSub.Udp.csproj" />
     <ProjectReference Include="..\..\Libraries\Opc.Ua.PubSub.Eth\Opc.Ua.PubSub.Eth.csproj" />
     <ProjectReference Include="..\..\Libraries\Opc.Ua.PubSub.Mqtt\Opc.Ua.PubSub.Mqtt.csproj" />
+    <ProjectReference Include="..\..\Libraries\Opc.Ua.PubSub.Kafka\Opc.Ua.PubSub.Kafka.csproj" />
     <ProjectReference Include="..\..\Libraries\Opc.Ua.PubSub.Adapter\Opc.Ua.PubSub.Adapter.csproj" />
   </ItemGroup>
   <ItemGroup>

Applications/ConsoleReferencePubSubClient/Program.cs

@@ -49,7 +49,7 @@ namespace Quickstarts.ConsoleReferencePubSubClient
     /// One executable exposes three command-line-selectable modes:
     /// <list type="bullet">
     /// <item><description>
-    /// <c>publisher</c> - publishes sample DataSets over UDP/UADP or MQTT (UADP/JSON).
+    /// <c>publisher</c> - publishes sample DataSets over UDP/UADP, Ethernet/UADP, MQTT, or Kafka (UADP/JSON).
     /// </description></item>
     /// <item><description>
     /// <c>subscriber</c> - receives DataSets and logs each decoded message.
@@ -88,14 +88,16 @@ public static async Task<int> Main(string[] args)
         }
 
         /// <summary>
-        /// Builds the <c>publisher</c> subcommand: a UDP/UADP or MQTT publisher
+        /// Builds the <c>publisher</c> subcommand: a UDP/UADP, Ethernet/UADP, MQTT, or Kafka publisher
         /// that publishes a built-in sample DataSet.
         /// </summary>
         private static Command BuildPublisherCommand(Action<int> setExitCode)
         {
             var profileOption = new Option<string>("--profile")
             {
-                Description = "Transport profile: udp-uadp | mqtt-uadp | mqtt-json.",
+                Description =
+                    "Transport profile: udp-uadp | eth-uadp | mqtt-uadp | mqtt-json | " +
+                    "kafka-uadp | kafka-json.",
                 DefaultValueFactory = _ => "udp-uadp"
             };
             var configFileOption = new Option<string?>("--config-file")
@@ -121,7 +123,8 @@ private static Command BuildPublisherCommand(Action<int> setExitCode)
             {
                 Description =
                     "Transport endpoint URL. Defaults: opc.udp://239.0.0.1:4840 (UDP), " +
-                    "mqtt://localhost:1883 (MQTT)."
+                    "opc.eth://01-00-5E-7F-00-01 (Ethernet), mqtt://localhost:1883 (MQTT), " +
+                    "kafka://localhost:9092 (Kafka)."
             };
             var intervalOption = new Option<int>("--interval")
             {
@@ -131,7 +134,7 @@ private static Command BuildPublisherCommand(Action<int> setExitCode)
 
             var command = new Command(
                 "publisher",
-                "Publish a sample DataSet over UDP/UADP or MQTT (UADP/JSON).")
+                "Publish a sample DataSet over UDP/UADP, Ethernet/UADP, MQTT, or Kafka (UADP/JSON).")
             {
                 profileOption,
                 configFileOption,
@@ -149,7 +152,7 @@ private static Command BuildPublisherCommand(Action<int> setExitCode)
                 {
                     await Console.Error.WriteLineAsync(
                         $"Unknown --profile value '{profileArg}'. " +
-                        "Expected one of: udp-uadp, mqtt-uadp, mqtt-json.")
+                        "Expected one of: udp-uadp, eth-uadp, mqtt-uadp, mqtt-json, kafka-uadp, kafka-json.")
                         .ConfigureAwait(false);
                     setExitCode(2);
                     return;
@@ -169,14 +172,16 @@ await Console.Error.WriteLineAsync(
         }
 
         /// <summary>
-        /// Builds the <c>subscriber</c> subcommand: a UDP/UADP or MQTT subscriber
+        /// Builds the <c>subscriber</c> subcommand: a UDP/UADP, Ethernet/UADP, MQTT, or Kafka subscriber
         /// that logs each decoded DataSetMessage.
         /// </summary>
         private static Command BuildSubscriberCommand(Action<int> setExitCode)
         {
             var profileOption = new Option<string>("--profile")
             {
-                Description = "Transport profile: udp-uadp | mqtt-uadp | mqtt-json.",
+                Description =
+                    "Transport profile: udp-uadp | eth-uadp | mqtt-uadp | mqtt-json | " +
+                    "kafka-uadp | kafka-json.",
                 DefaultValueFactory = _ => "udp-uadp"
             };
             var configFileOption = new Option<string?>("--config-file")
@@ -200,8 +205,10 @@ private static Command BuildSubscriberCommand(Action<int> setExitCode)
             };
             var endpointOption = new Option<string?>("--endpoint")
             {
-                Description = "Transport endpoint URL. Defaults: opc.udp://239.0.0.1:4840 " +
-                    "(UDP), mqtt://localhost:1883 (MQTT)."
+                Description =
+                    "Transport endpoint URL. Defaults: opc.udp://239.0.0.1:4840 (UDP), " +
+                    "opc.eth://01-00-5E-7F-00-01 (Ethernet), mqtt://localhost:1883 (MQTT), " +
+                    "kafka://localhost:9092 (Kafka)."
             };
 
             var command = new Command(
@@ -223,7 +230,7 @@ private static Command BuildSubscriberCommand(Action<int> setExitCode)
                 {
                     await Console.Error.WriteLineAsync(
                         $"Unknown --profile value '{profileArg}'. " +
-                        "Expected one of: udp-uadp, mqtt-uadp, mqtt-json.")
+                        "Expected one of: udp-uadp, eth-uadp, mqtt-uadp, mqtt-json, kafka-uadp, kafka-json.")
                         .ConfigureAwait(false);
                     setExitCode(2);
                     return;
@@ -373,13 +380,23 @@ private static async Task<int> RunPublisherAsync(
                     .AddUdpTransport()
                     .AddSecurityKeyProvider(SampleSecurity.CreateKeyProvider())
                     .AddDataSetSource(PublisherConfigurationBuilder.DataSetName, sampleSource);
-                if (profile == PublisherProfile.EthUadp)
+                switch (profile)
                 {
-                    publisher.AddEthTransport();
-                }
-                else if (profile != PublisherProfile.UdpUadp)
-                {
-                    publisher.AddMqttTransport();
+                    case PublisherProfile.UdpUadp:
+                        break;
+                    case PublisherProfile.EthUadp:
+                        publisher.AddEthTransport();
+                        break;
+                    case PublisherProfile.MqttUadp:
+                    case PublisherProfile.MqttJson:
+                        publisher.AddMqttTransport();
+                        break;
+                    case PublisherProfile.KafkaUadp:
+                    case PublisherProfile.KafkaJson:
+                        publisher.AddKafkaTransport();
+                        break;
+                    default:
+                        throw new ArgumentOutOfRangeException(nameof(profile), profile, null);
                 }
                 publisher.ConfigureApplication(app =>
                 {
@@ -441,13 +458,23 @@ private static async Task<int> RunSubscriberAsync(
                         sp => new ConsoleLoggingSink(
                             sp.GetRequiredService<ILoggerFactory>()
                                 .CreateLogger<ConsoleLoggingSink>()));
-                if (profile == SubscriberProfile.EthUadp)
-                {
-                    subscriber.AddEthTransport();
-                }
-                else if (profile != SubscriberProfile.UdpUadp)
+                switch (profile)
                 {
-                    subscriber.AddMqttTransport();
+                    case SubscriberProfile.UdpUadp:
+                        break;
+                    case SubscriberProfile.EthUadp:
+                        subscriber.AddEthTransport();
+                        break;
+                    case SubscriberProfile.MqttUadp:
+                    case SubscriberProfile.MqttJson:
+                        subscriber.AddMqttTransport();
+                        break;
+                    case SubscriberProfile.KafkaUadp:
+                    case SubscriberProfile.KafkaJson:
+                        subscriber.AddKafkaTransport();
+                        break;
+                    default:
+                        throw new ArgumentOutOfRangeException(nameof(profile), profile, null);
                 }
                 subscriber.ConfigureApplication(app =>
                 {
@@ -719,6 +746,12 @@ private static bool TryParsePublisherProfile(string? text, out PublisherProfile
                 case "eth-uadp":
                     profile = PublisherProfile.EthUadp;
                     return true;
+                case "kafka-uadp":
+                    profile = PublisherProfile.KafkaUadp;
+                    return true;
+                case "kafka-json":
+                    profile = PublisherProfile.KafkaJson;
+                    return true;
                 default:
                     profile = PublisherProfile.UdpUadp;
                     return false;
@@ -741,6 +774,12 @@ private static bool TryParseSubscriberProfile(string? text, out SubscriberProfil
                 case "eth-uadp":
                     profile = SubscriberProfile.EthUadp;
                     return true;
+                case "kafka-uadp":
+                    profile = SubscriberProfile.KafkaUadp;
+                    return true;
+                case "kafka-json":
+                    profile = SubscriberProfile.KafkaJson;
+                    return true;
                 default:
                     profile = SubscriberProfile.UdpUadp;
                     return false;
@@ -843,7 +882,17 @@ public enum PublisherProfile
         /// <summary>
         /// Ethernet (Layer 2) transport with UADP message mapping.
         /// </summary>
-        EthUadp = 3
+        EthUadp = 3,
+
+        /// <summary>
+        /// Kafka broker transport with UADP message mapping.
+        /// </summary>
+        KafkaUadp = 4,
+
+        /// <summary>
+        /// Kafka broker transport with JSON message mapping.
+        /// </summary>
+        KafkaJson = 5
     }
 
     /// <summary>
@@ -869,7 +918,17 @@ public enum SubscriberProfile
         /// <summary>
         /// Ethernet (Layer 2) transport with UADP message mapping.
         /// </summary>
-        EthUadp = 3
+        EthUadp = 3,
+
+        /// <summary>
+        /// Kafka broker transport with UADP message mapping.
+        /// </summary>
+        KafkaUadp = 4,
+
+        /// <summary>
+        /// Kafka broker transport with JSON message mapping.
+        /// </summary>
+        KafkaJson = 5
     }
 
     /// <summary>

Applications/ConsoleReferencePubSubClient/PublisherConfigurationBuilder.cs

@@ -31,12 +31,13 @@
 using Opc.Ua;
 using Opc.Ua.PubSub.Configuration;
 using Opc.Ua.PubSub.Eth;
+using Opc.Ua.PubSub.Kafka;
 
 namespace Quickstarts.ConsoleReferencePubSubClient
 {
     /// <summary>
     /// Builds minimal Part 14 <see cref="PubSubConfigurationDataType"/>
-    /// payloads for the three demo wire profiles using the fluent
+    /// payloads for the demo wire profiles using the fluent
     /// <see cref="PubSubConfigurationBuilder"/>. The payloads use the
     /// "Simple" DataSet exposed by <see cref="SampleDataSetSource"/>
     /// (BoolToggle, Int32 counter, DateTime).
@@ -47,15 +48,19 @@ public static class PublisherConfigurationBuilder
         public const string DefaultUdpEndpoint = "opc.udp://239.0.0.1:4840";
         public const string DefaultEthEndpoint = "opc.eth://01-00-5E-7F-00-01";
         public const string DefaultMqttEndpoint = "mqtt://localhost:1883";
+        public const string DefaultKafkaEndpoint = "kafka://localhost:9092";
         private const string MqttQueueName = "Quickstarts/Reference/Simple";
+        private const string KafkaTopicName = "Quickstarts.Reference.Simple";
 
         public static string DefaultEndpointFor(PublisherProfile profile)
         {
             return profile switch
             {
                 PublisherProfile.UdpUadp => DefaultUdpEndpoint,
                 PublisherProfile.EthUadp => DefaultEthEndpoint,
-                _ => DefaultMqttEndpoint
+                PublisherProfile.KafkaUadp or PublisherProfile.KafkaJson => DefaultKafkaEndpoint,
+                PublisherProfile.MqttUadp or PublisherProfile.MqttJson => DefaultMqttEndpoint,
+                _ => throw new ArgumentOutOfRangeException(nameof(profile))
             };
         }
 
@@ -68,20 +73,25 @@ public static PubSubConfigurationDataType Build(
             int intervalMs)
         {
             // UDP and Ethernet are datagram transports (no broker queue);
-            // the MQTT profiles use broker transport settings instead.
+            // broker profiles use broker transport settings instead.
             bool udp = profile is PublisherProfile.UdpUadp or PublisherProfile.EthUadp;
 
             // UADP message security (SignAndEncrypt) is wired for the UADP
             // profiles via the shared StaticSecurityKeyProvider. The JSON
             // profile has no UADP security wrapper, so it stays unsecured.
-            bool secured = profile != PublisherProfile.MqttJson;
+            bool secured = profile is not (PublisherProfile.MqttJson or PublisherProfile.KafkaJson);
+            string brokerQueueName = profile is PublisherProfile.KafkaUadp or PublisherProfile.KafkaJson
+                ? KafkaTopicName
+                : MqttQueueName;
 
             string transportProfileUri = profile switch
             {
                 PublisherProfile.UdpUadp => Profiles.PubSubUdpUadpTransport,
                 PublisherProfile.EthUadp => EthProfiles.PubSubEthUadpTransport,
                 PublisherProfile.MqttUadp => Profiles.PubSubMqttUadpTransport,
                 PublisherProfile.MqttJson => Profiles.PubSubMqttJsonTransport,
+                PublisherProfile.KafkaUadp => KafkaProfiles.PubSubKafkaUadpTransport,
+                PublisherProfile.KafkaJson => KafkaProfiles.PubSubKafkaJsonTransport,
                 _ => throw new ArgumentOutOfRangeException(nameof(profile))
             };
 
@@ -104,7 +114,7 @@ public static PubSubConfigurationDataType Build(
                                     ? new DatagramWriterGroupTransportDataType()
                                     : new BrokerWriterGroupTransportDataType
                                     {
-                                        QueueName = MqttQueueName
+                                        QueueName = brokerQueueName
                                     });
                             if (secured)
                             {
@@ -126,7 +136,7 @@ public static PubSubConfigurationDataType Build(
                                     writer.WithTransportSettings(
                                         new BrokerDataSetWriterTransportDataType
                                         {
-                                            QueueName = MqttQueueName,
+                                            QueueName = brokerQueueName,
                                             RequestedDeliveryGuarantee
                                                 = BrokerTransportQualityOfService.BestEffort
                                         });
@@ -138,7 +148,7 @@ public static PubSubConfigurationDataType Build(
 
         private static IEncodeable WriterGroupMessageSettings(PublisherProfile profile)
         {
-            if (profile == PublisherProfile.MqttJson)
+            if (profile is PublisherProfile.MqttJson or PublisherProfile.KafkaJson)
             {
                 return new JsonWriterGroupMessageDataType
                 {
@@ -163,7 +173,7 @@ private static IEncodeable WriterGroupMessageSettings(PublisherProfile profile)
 
         private static IEncodeable WriterMessageSettings(PublisherProfile profile)
         {
-            if (profile == PublisherProfile.MqttJson)
+            if (profile is PublisherProfile.MqttJson or PublisherProfile.KafkaJson)
             {
                 return new JsonDataSetWriterMessageDataType
                 {