diff --git a/Directory.Packages.props b/Directory.Packages.props index 28f9da30b4..254bbc830d 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -16,7 +16,7 @@ - + @@ -38,7 +38,7 @@ - + @@ -49,16 +49,16 @@ - + - + - + @@ -108,7 +108,7 @@ - + diff --git a/docs/PubSub.md b/docs/PubSub.md index d0882c8faf..2845f335fb 100644 --- a/docs/PubSub.md +++ b/docs/PubSub.md @@ -101,7 +101,7 @@ Actions to an external OPC UA server over a managed client session. │ │ │ ┌─────────────────┐ ┌──────────────────────┐ ┌────────────────────┐ │ Opc.Ua.PubSub. │ │ Opc.Ua.PubSub.Mqtt │ │ Opc.Ua.PubSub.Kafka│ -│ Udp │ │ MQTTnet 4 / 5 │ │ Dekaf / librdkafka │ +│ Udp │ │ MQTTnet 4 / 5 │ │ Dekaf / Confluent │ └─────────────────┘ └──────────────────────┘ └────────────────────┘ ``` @@ -756,6 +756,15 @@ pubsub.AddPublisher() }); ``` +The managed Dekaf client is the default on every target framework. JIT-compiled hosts can select the native Confluent.Kafka alternative through DI: + +```csharp +pubsub.AddKafkaTransport() + .WithConfluentKafkaClient(); +``` + +The Confluent backend uses native librdkafka and is not NativeAOT or trimming compatible. + Subscribers set `GroupId` (consumer group) and `AutoOffsetReset` instead of the delivery guarantee. Writer and reader groups use the same fluent `PubSubConfigurationBuilder` shown under [Fluent builder walkthrough](#fluent-builder-walkthrough); the [reference sample](../samples/ConsoleReferencePubSubClient/README.md) contains complete Kafka publisher and subscriber configurations. **Topic mapping.** Kafka topics come from the OPC UA broker transport settings: `BrokerDataSetWriterTransportDataType.QueueName` / `BrokerDataSetReaderTransportDataType.QueueName` select the per-writer/reader data topic, `BrokerWriterGroupTransportDataType.QueueName` is the writer-group fallback, and `MetaDataQueueName` selects the metadata topic. When `MetaDataQueueName` is unset the transport derives a deterministic fallback from `KafkaConnectionOptions.Topics.Prefix`, the encoding, message type, PublisherId, WriterGroupId, and DataSetWriterId (segments joined with `.`). Use Kafka-safe characters (letters, digits, `.`, `_`, `-`). @@ -764,7 +773,7 @@ Subscribers set `GroupId` (consumer group) and `AutoOffsetReset` instead of the **SASL and TLS.** Use `kafkas://` or `KafkaConnectionOptions.Tls.UseTls` for TLS; `KafkaTlsOptions` carries CA / client-certificate / client-key PEM paths. `SecurityProtocol = KafkaSecurityProtocol.SaslSsl` with `SaslMechanism`, `UserName`, and `PasswordSecretId` enables SASL over TLS, and `PasswordSecretId` is resolved through the OPC UA secret store so configuration never carries a plaintext password. Sending SASL credentials over plaintext `kafka://` is rejected unless `AllowCredentialsOverPlaintext` is explicitly set for local development. -**NativeAOT.** On `net10.0` the transport uses the pure-managed [Dekaf](https://github.com/thomhurst/Dekaf) client and is NativeAOT/trimming compatible; on `net472`, `net48`, `netstandard2.1`, `net8.0`, and `net9.0` it uses `Confluent.Kafka` (native librdkafka), which is not AOT-compatible. See [Native AOT](#native-aot). +**NativeAOT.** The default pure-managed [Dekaf](https://github.com/thomhurst/Dekaf) backend is validated for NativeAOT/trimming on `net10.0`. The opt-in Confluent.Kafka backend uses native librdkafka and is not NativeAOT or trimming compatible. See [Native AOT](#native-aot). ## Encodings @@ -1826,11 +1835,7 @@ detailed the counters become; configure via ## Native AOT -The four core PubSub assemblies (`Opc.Ua.PubSub`, `.Udp`, `.Eth`, `.Mqtt`) are -AOT-clean on every target framework. The `Opc.Ua.PubSub.Kafka` transport is -AOT-clean only on `net10.0` (managed Dekaf client); on the other frameworks it -uses `Confluent.Kafka` / native librdkafka and is not AOT-compatible (see -[Apache Kafka](#apache-kafka)). +The PubSub assemblies (`Opc.Ua.PubSub`, `.Udp`, `.Eth`, `.Mqtt`, and `.Kafka`) are AOT-clean on the repository's NativeAOT validation target (`net10.0`) when using their default backends. The Kafka transport's opt-in Confluent.Kafka backend uses native librdkafka and is not AOT-compatible (see [Apache Kafka](#apache-kafka)). - **No reflection-based serialization.** Source-generated `IEncodeable` types (Part 14 datatypes) plus hand-written diff --git a/src/Opc.Ua.PubSub.Kafka/DependencyInjection/KafkaTransportServiceCollectionExtensions.cs b/src/Opc.Ua.PubSub.Kafka/DependencyInjection/KafkaTransportServiceCollectionExtensions.cs index 45a7e374d7..b5a3d5c5f8 100644 --- a/src/Opc.Ua.PubSub.Kafka/DependencyInjection/KafkaTransportServiceCollectionExtensions.cs +++ b/src/Opc.Ua.PubSub.Kafka/DependencyInjection/KafkaTransportServiceCollectionExtensions.cs @@ -139,6 +139,27 @@ public static IPubSubBuilder AddKafkaTransport( return builder; } + /// + /// Replaces the default managed Dekaf client with Confluent.Kafka. + /// + /// + /// Confluent.Kafka uses native librdkafka and is not NativeAOT or + /// trimming compatible. Use this option only for JIT-compiled hosts. + /// + /// PubSub builder. + /// The same for chaining. + /// + public static IPubSubBuilder WithConfluentKafkaClient(this IPubSubBuilder builder) + { + if (builder is null) + { + throw new ArgumentNullException(nameof(builder)); + } + builder.Services.Replace( + ServiceDescriptor.Singleton()); + return builder; + } + /// /// One-shot: registers a PubSub publisher and subscriber together with /// the Apache Kafka transport (JSON + UADP profiles) on a fresh OPC UA @@ -187,11 +208,7 @@ public static IOpcUaBuilder AddKafkaPubSub( private static void RegisterShared(IServiceCollection services) { -#if NET10_0_OR_GREATER services.TryAddSingleton(); -#else - services.TryAddSingleton(); -#endif services.AddPubSubTransportFactory(sp => new KafkaPubSubTransportFactory( KafkaProfiles.PubSubKafkaJsonTransport, diff --git a/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientAdapter.cs b/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientAdapter.cs index 78278131bd..c6da48d03d 100644 --- a/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientAdapter.cs +++ b/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientAdapter.cs @@ -27,7 +27,6 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ -#if !NET10_0_OR_GREATER using System; using System.Collections.Generic; using System.Globalization; @@ -49,8 +48,8 @@ namespace Opc.Ua.PubSub.Kafka.Internal /// loop (built on first subscribe), so a send-only or receive-only /// connection never instantiates the unused half. Confluent.Kafka /// wraps native librdkafka via P/Invoke and is therefore not - /// NativeAOT/trim-safe; this transport library is intentionally - /// excluded from AOT publishing (see the project file). + /// NativeAOT/trim-safe. Select it only for JIT-compiled hosts through + /// WithConfluentKafkaClient(). /// internal sealed class ConfluentKafkaClientAdapter : IKafkaClientAdapter { @@ -652,6 +651,4 @@ public static partial void KafkaConsumeError( Message = "Kafka error {Code}: {Reason}")] public static partial void KafkaError(this ILogger logger, ErrorCode code, string? reason); } - } -#endif diff --git a/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientFactory.cs b/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientFactory.cs index 871361ca38..e48f5e9939 100644 --- a/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientFactory.cs +++ b/src/Opc.Ua.PubSub.Kafka/Internal/ConfluentKafkaClientFactory.cs @@ -28,19 +28,17 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ -#if !NET10_0_OR_GREATER using System; namespace Opc.Ua.PubSub.Kafka.Internal { /// - /// Default implementation backed by + /// Optional implementation backed by /// Confluent.Kafka (native librdkafka). /// /// - /// Wired into the DI container by the PubSub transport composition; - /// tests may instantiate it directly or substitute a fake factory to - /// avoid an actual broker connection. + /// Selected through WithConfluentKafkaClient(). The default + /// transport registration uses the managed Dekaf client. /// internal sealed class ConfluentKafkaClientFactory : IKafkaClientFactory { @@ -59,4 +57,3 @@ public IKafkaClientAdapter Create(ITelemetryContext telemetry, TimeProvider time } } } -#endif diff --git a/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientAdapter.cs b/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientAdapter.cs index 8f9fa278b6..18d31ed248 100644 --- a/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientAdapter.cs +++ b/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientAdapter.cs @@ -27,10 +27,12 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ -#if NET10_0_OR_GREATER using System; using System.Collections.Generic; using System.Globalization; +#if NETFRAMEWORK +using System.Runtime.InteropServices; +#endif using System.Threading; using System.Threading.Tasks; using Dekaf; @@ -51,7 +53,7 @@ namespace Opc.Ua.PubSub.Kafka.Internal /// plus a background consume /// loop (built on first subscribe), so a send-only or receive-only /// connection never instantiates the unused half. Dekaf is a pure-managed - /// .NET 10 Kafka client and is therefore usable by NativeAOT publishers. + /// Kafka client and is therefore usable by NativeAOT publishers. /// internal sealed class DekafKafkaClientAdapter : IKafkaClientAdapter { @@ -447,9 +449,32 @@ private void DispatchRecord(ConsumeResult result) { foreach (Header header in headers) { - string headerValue = header.IsValueNull - ? string.Empty - : System.Text.Encoding.UTF8.GetString(header.Value.Span); + string headerValue; + if (header.IsValueNull) + { + headerValue = string.Empty; + } +#if NETFRAMEWORK + else if (MemoryMarshal.TryGetArray( + header.Value, + out ArraySegment segment) && + segment.Array is not null) + { + headerValue = System.Text.Encoding.UTF8.GetString( + segment.Array, + segment.Offset, + segment.Count); + } + else + { + headerValue = System.Text.Encoding.UTF8.GetString(header.Value.ToArray()); + } +#else + else + { + headerValue = System.Text.Encoding.UTF8.GetString(header.Value.Span); + } +#endif if (string.Equals(header.Key, ContentTypeHeader, StringComparison.OrdinalIgnoreCase)) { contentType = headerValue; @@ -724,4 +749,3 @@ private static NotSupportedException CreateUnsupportedSaslMechanismException(Kaf } } } -#endif diff --git a/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientFactory.cs b/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientFactory.cs index 729c237dcd..ee668e8ba2 100644 --- a/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientFactory.cs +++ b/src/Opc.Ua.PubSub.Kafka/Internal/DekafKafkaClientFactory.cs @@ -28,7 +28,6 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ -#if NET10_0_OR_GREATER using System; namespace Opc.Ua.PubSub.Kafka.Internal @@ -59,4 +58,3 @@ public IKafkaClientAdapter Create(ITelemetryContext telemetry, TimeProvider time } } } -#endif diff --git a/src/Opc.Ua.PubSub.Kafka/Internal/IKafkaClientFactory.cs b/src/Opc.Ua.PubSub.Kafka/Internal/IKafkaClientFactory.cs index b9558b1516..747342e4a8 100644 --- a/src/Opc.Ua.PubSub.Kafka/Internal/IKafkaClientFactory.cs +++ b/src/Opc.Ua.PubSub.Kafka/Internal/IKafkaClientFactory.cs @@ -35,7 +35,8 @@ namespace Opc.Ua.PubSub.Kafka.Internal /// Provider-model factory for the Kafka client adapter used by /// . Test code can swap in a fake /// to drive the transport without an actual broker; the default - /// implementation creates a TFM-specific Kafka-backed adapter. + /// implementation creates a managed Dekaf adapter and DI can select + /// the Confluent.Kafka alternative. /// /// /// Provides the adapter seam used by the Kafka broker transport per diff --git a/src/Opc.Ua.PubSub.Kafka/KafkaBrokerTransport.cs b/src/Opc.Ua.PubSub.Kafka/KafkaBrokerTransport.cs index 11883bf94a..833b949997 100644 --- a/src/Opc.Ua.PubSub.Kafka/KafkaBrokerTransport.cs +++ b/src/Opc.Ua.PubSub.Kafka/KafkaBrokerTransport.cs @@ -66,11 +66,11 @@ namespace Opc.Ua.PubSub.Kafka /// /// /// The transport delegates to an so - /// the Confluent.Kafka client surface is invisible to higher layers, - /// and so unit tests can inject a fake adapter to exercise the state - /// machine without an actual broker. Each produced record carries a - /// partition key derived from the PublisherId so records from a - /// publisher preserve ordering within a partition. + /// the Kafka client implementation is invisible to higher layers, and + /// so unit tests can inject a fake adapter to exercise the state machine + /// without an actual broker. Each produced record carries a partition + /// key derived from the PublisherId so records from a publisher preserve + /// ordering within a partition. /// /// public sealed class KafkaBrokerTransport diff --git a/src/Opc.Ua.PubSub.Kafka/KafkaConnectionOptions.cs b/src/Opc.Ua.PubSub.Kafka/KafkaConnectionOptions.cs index c309f935c7..0c29fccc0f 100644 --- a/src/Opc.Ua.PubSub.Kafka/KafkaConnectionOptions.cs +++ b/src/Opc.Ua.PubSub.Kafka/KafkaConnectionOptions.cs @@ -33,8 +33,8 @@ namespace Opc.Ua.PubSub.Kafka { /// /// Transport-level security protocol negotiated with the Kafka - /// brokers, mirroring the librdkafka security.protocol - /// property. + /// brokers, corresponding to the standard security.protocol + /// setting. /// /// /// Implements the security profile selector of @@ -67,8 +67,8 @@ public enum KafkaSecurityProtocol /// /// SASL authentication mechanism used when the negotiated - /// carries SASL, mirroring the - /// librdkafka sasl.mechanism property. + /// carries SASL, corresponding to + /// the standard sasl.mechanism setting. /// /// /// Implements the SASL mechanism selector of @@ -105,8 +105,8 @@ public enum KafkaSaslMechanism /// /// Offset reset policy applied when a consumer group has no - /// committed offset for a partition, mirroring the librdkafka - /// auto.offset.reset property. + /// committed offset for a partition, corresponding to the standard + /// auto.offset.reset setting. /// /// /// Implements the subscriber start-offset policy of @@ -160,9 +160,8 @@ public sealed class KafkaConnectionOptions /// /// Comma-separated host:port bootstrap server list passed - /// to librdkafka bootstrap.servers. Populated by the - /// transport factory from when not set - /// explicitly. + /// to the Kafka client. Populated by the transport factory from + /// when not set explicitly. /// public string BootstrapServers { get; set; } = string.Empty; @@ -235,8 +234,8 @@ public sealed class KafkaConnectionOptions public KafkaTlsOptions? Tls { get; set; } /// - /// Producer delivery guarantee mapped to the librdkafka - /// acks / enable.idempotence settings. Defaults to + /// Producer delivery guarantee mapped to the Kafka acks + /// and idempotent-producer settings. Defaults to /// . /// public KafkaQualityOfService DeliveryGuarantee { get; set; } @@ -268,15 +267,13 @@ public sealed class KafkaConnectionOptions /// /// Maximum time the producer waits for a record to be delivered - /// before failing it, mapped to librdkafka - /// message.timeout.ms. + /// before failing it. /// public TimeSpan MessageTimeout { get; set; } = TimeSpan.FromSeconds(30); /// - /// Maximum size (in bytes) of a single Kafka record, mapped to - /// librdkafka message.max.bytes. The default of 1048576 - /// matches the common broker default. + /// Maximum configured size (in bytes) of a single Kafka record. + /// The default of 1048576 matches the common broker default. /// public int MaxMessageSize { get; set; } = 1048576; diff --git a/src/Opc.Ua.PubSub.Kafka/KafkaQualityOfService.cs b/src/Opc.Ua.PubSub.Kafka/KafkaQualityOfService.cs index 299cbc9ca7..8cf27a833d 100644 --- a/src/Opc.Ua.PubSub.Kafka/KafkaQualityOfService.cs +++ b/src/Opc.Ua.PubSub.Kafka/KafkaQualityOfService.cs @@ -80,8 +80,7 @@ public enum KafkaQualityOfService /// /// Broker acknowledgement level requested from the Kafka producer. - /// Numeric values match the librdkafka acks wire encoding so - /// the adapter can map without an extra lookup. + /// Numeric values match the Kafka acks setting. /// /// /// Backs the acks selector described in @@ -115,8 +114,8 @@ public enum KafkaAcks /// /// /// Carries the two producer knobs that realise the Part 14 Annex B.2 - /// delivery guarantee so the Confluent-backed adapter can apply them - /// without re-deriving the mapping. + /// delivery guarantee so Kafka client adapters can apply them without + /// re-deriving the mapping. /// /// Broker acknowledgement level. /// diff --git a/src/Opc.Ua.PubSub.Kafka/KafkaTlsOptions.cs b/src/Opc.Ua.PubSub.Kafka/KafkaTlsOptions.cs index 3b7af0030a..16271ccad3 100644 --- a/src/Opc.Ua.PubSub.Kafka/KafkaTlsOptions.cs +++ b/src/Opc.Ua.PubSub.Kafka/KafkaTlsOptions.cs @@ -39,10 +39,10 @@ namespace Opc.Ua.PubSub.Kafka /// Backs the TLS transport surface required by /// /// Part 14 Annex B.2 Apache Kafka transport. Unlike the MQTT - /// transport, the underlying librdkafka client consumes certificate - /// and key material through file-system paths rather than the OPC UA - /// certificate store, so this POCO references PEM file locations. No - /// private key material is embedded in configuration files. + /// transport, the Kafka client consumes certificate and key material + /// through file-system paths rather than the OPC UA certificate store, + /// so this POCO references PEM file locations. No private key material + /// is embedded in configuration files. /// public sealed class KafkaTlsOptions { @@ -66,23 +66,20 @@ public sealed class KafkaTlsOptions /// /// Path to a PEM file containing the certificate authority (CA) /// certificates that form the trust chain used to validate the - /// broker certificate. Maps to the librdkafka - /// ssl.ca.location property. defers - /// to the platform / runtime default trust store. + /// broker certificate. defers to the + /// platform / runtime default trust store. /// public string? CaCertificatePath { get; set; } /// /// Path to a PEM file containing the client certificate presented - /// during the TLS handshake for mutual TLS. Maps to the - /// librdkafka ssl.certificate.location property. + /// during the TLS handshake for mutual TLS. /// public string? ClientCertificatePath { get; set; } /// /// Path to a PEM file containing the client private key that - /// matches . Maps to the - /// librdkafka ssl.key.location property. + /// matches . /// public string? ClientKeyPath { get; set; } diff --git a/src/Opc.Ua.PubSub.Kafka/NugetREADME.md b/src/Opc.Ua.PubSub.Kafka/NugetREADME.md index 242321adb7..a1020f2de6 100644 --- a/src/Opc.Ua.PubSub.Kafka/NugetREADME.md +++ b/src/Opc.Ua.PubSub.Kafka/NugetREADME.md @@ -1,10 +1,6 @@ # OPC UA .NET Standard — PubSub Apache Kafka transport -`OPCFoundation.NetStandard.Opc.Ua.PubSub.Kafka` provides the Apache Kafka broker -transport (OPC UA Part 14 Annex B.2, with SASL/TLS security, configurable delivery -guarantees, and both the UADP and JSON message mappings) for the modern -`OPCFoundation.NetStandard.Opc.Ua.PubSub` stack. Kafka consumer groups and idempotent -producers back the high-availability publisher/subscriber deployments. +`OPCFoundation.NetStandard.Opc.Ua.PubSub.Kafka` provides the Apache Kafka broker transport (OPC UA Part 14 Annex B.2, with SASL/TLS security, configurable delivery guarantees, and both the UADP and JSON message mappings) for the modern `OPCFoundation.NetStandard.Opc.Ua.PubSub` stack. Kafka consumer groups and idempotent producers back the high-availability publisher/subscriber deployments. ## Getting started @@ -27,13 +23,16 @@ Connection addresses use `kafka://host:9092` (plain/SASL) or `kafkas://host:9093 ## NativeAOT -The Kafka client is dual-sourced by target framework: +The transport defaults to the pure-managed [Dekaf](https://github.com/thomhurst/Dekaf) client on every supported target framework. This repository asserts and validates **NativeAOT / trimming compatibility on `net10.0`** for that default backend. -- On **net10.0** the transport uses the pure-managed [Dekaf](https://github.com/thomhurst/Dekaf) - client (no native dependency) and is **NativeAOT / trimming compatible**. -- On **net472, net48, netstandard2.1, net8.0, and net9.0** it uses `Confluent.Kafka` - (native `librdkafka`), which is **not** NativeAOT/trimming compatible — use a JIT-compiled - host on those frameworks. +JIT-compiled hosts can opt into `Confluent.Kafka`: + +```csharp +pubsub.AddKafkaTransport() + .WithConfluentKafkaClient(); +``` + +The Confluent backend uses native `librdkafka` and is not NativeAOT or trimming compatible. The other PubSub transports (UDP, Ethernet, MQTT) remain AOT-compatible on all frameworks. diff --git a/src/Opc.Ua.PubSub.Kafka/Opc.Ua.PubSub.Kafka.csproj b/src/Opc.Ua.PubSub.Kafka/Opc.Ua.PubSub.Kafka.csproj index 96b645edd9..c0de6f81d1 100644 --- a/src/Opc.Ua.PubSub.Kafka/Opc.Ua.PubSub.Kafka.csproj +++ b/src/Opc.Ua.PubSub.Kafka/Opc.Ua.PubSub.Kafka.csproj @@ -11,12 +11,7 @@ enable $(NoWarn);CS1591 true - + true @@ -32,12 +27,9 @@ - - - - - + + diff --git a/tests/Opc.Ua.Aot.Tests/Opc.Ua.Aot.Tests.csproj b/tests/Opc.Ua.Aot.Tests/Opc.Ua.Aot.Tests.csproj index 8d3f890d39..79127a9591 100644 --- a/tests/Opc.Ua.Aot.Tests/Opc.Ua.Aot.Tests.csproj +++ b/tests/Opc.Ua.Aot.Tests/Opc.Ua.Aot.Tests.csproj @@ -8,7 +8,7 @@ disable $(NoWarn);IL2104;IL3053 diff --git a/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaClientAdapterGuardTests.cs b/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaClientAdapterGuardTests.cs index 91663f382a..a9c7ccdad7 100644 --- a/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaClientAdapterGuardTests.cs +++ b/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaClientAdapterGuardTests.cs @@ -27,12 +27,13 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ -#if NET10_0_OR_GREATER using System; using System.Collections.Generic; using System.Reflection; using System.Threading; using System.Threading.Tasks; +using Dekaf.Consumer; +using Dekaf.Serialization; using NUnit.Framework; using Opc.Ua.PubSub.Kafka.Internal; using Opc.Ua.PubSub.Tests; @@ -41,11 +42,11 @@ namespace Opc.Ua.PubSub.Kafka.Tests { /// - /// Guard tests for the default Dekaf-backed Kafka adapter that do not require a broker. + /// Guard tests for Kafka client adapters that do not require a broker. /// [TestFixture] [Category("Unit")] - [TestSpec("B.2", Summary = "Kafka default adapter guard behavior")] + [TestSpec("B.2", Summary = "Kafka client adapter guard behavior")] [CancelAfter(10000)] public sealed class KafkaClientAdapterGuardTests { @@ -58,16 +59,28 @@ public void ConstructorRejectsInvalidArguments() Assert.That( () => new DekafKafkaClientAdapter(NUnitTelemetryContext.Create(), null!), Throws.TypeOf()); + Assert.That( + () => new ConfluentKafkaClientAdapter(null!, TimeProvider.System), + Throws.TypeOf()); + Assert.That( + () => new ConfluentKafkaClientAdapter(NUnitTelemetryContext.Create(), null!), + Throws.TypeOf()); } [Test] - public void FactoryCreatesDekafAdapter() + public void FactoriesCreateRequestedAdapters() { - var factory = new DekafKafkaClientFactory(); + var dekafFactory = new DekafKafkaClientFactory(); + var confluentFactory = new ConfluentKafkaClientFactory(); - object adapter = factory.Create(NUnitTelemetryContext.Create(), TimeProvider.System); + object dekafAdapter = dekafFactory.Create(NUnitTelemetryContext.Create(), TimeProvider.System); + object confluentAdapter = confluentFactory.Create(NUnitTelemetryContext.Create(), TimeProvider.System); - Assert.That(adapter, Is.InstanceOf()); + Assert.Multiple(() => + { + Assert.That(dekafAdapter, Is.InstanceOf()); + Assert.That(confluentAdapter, Is.InstanceOf()); + }); } [Test] @@ -156,6 +169,42 @@ await adapter.ConnectAsync(new KafkaConnectionOptions { Endpoint = KafkaTestHelp Throws.TypeOf()); } + [Test] + public async Task DispatchRecordWithNullHeaderMapsEmptyValueAsync() + { + await using var adapter = new DekafKafkaClientAdapter( + NUnitTelemetryContext.Create(), + TimeProvider.System); + KafkaIncomingMessageEventArgs? received = null; + adapter.IncomingMessage += (_, args) => received = args; + + InvokeDispatchRecord(adapter, CreateConsumeResult(new Header("x-null", (byte[]?)null))); + + Assert.That(received, Is.Not.Null); + Assert.That(received!.Message.Headers, Is.Not.Null); + Assert.That(received.Message.Headers!["x-null"], Is.Empty); + } + + [Test] + public async Task DispatchRecordWithUtf8HeaderDecodesValueAsync() + { + await using var adapter = new DekafKafkaClientAdapter( + NUnitTelemetryContext.Create(), + TimeProvider.System); + KafkaIncomingMessageEventArgs? received = null; + adapter.IncomingMessage += (_, args) => received = args; + const string headerValue = "Gr\u00FC\u00DFe"; + + InvokeDispatchRecord( + adapter, + CreateConsumeResult( + new Header("x-text", System.Text.Encoding.UTF8.GetBytes(headerValue)))); + + Assert.That(received, Is.Not.Null); + Assert.That(received!.Message.Headers, Is.Not.Null); + Assert.That(received.Message.Headers!["x-text"], Is.EqualTo(headerValue)); + } + [Test] public void PrivateMappingHelpersCoverKafkaConfigurationBranches() { @@ -285,6 +334,41 @@ public void PrivateProducerAndConsumerConfigHelpersCoverSecurityBranches() Throws.TypeOf()); } + private static ConsumeResult CreateConsumeResult(params Header[] headers) + { + return new ConsumeResult( + topic: KafkaTestHelper.JsonTopic, + partition: 0, + offset: 1, + keyData: ReadOnlyMemory.Empty, + isKeyNull: true, + valueData: ReadOnlyMemory.Empty, + isValueNull: true, + headers: headers, + timestampMs: 0, + timestampType: TimestampType.NotAvailable, + leaderEpoch: null, + keyDeserializer: null, + valueDeserializer: null); + } + + private static void InvokeDispatchRecord( + DekafKafkaClientAdapter adapter, + ConsumeResult result) + { + MethodInfo method = typeof(DekafKafkaClientAdapter).GetMethod( + "DispatchRecord", + BindingFlags.NonPublic | BindingFlags.Instance)!; + try + { + method.Invoke(adapter, [result]); + } + catch (TargetInvocationException ex) when (ex.InnerException is not null) + { + throw ex.InnerException; + } + } + private static T InvokePrivate(string methodName, params object?[] args) { MethodInfo method = typeof(DekafKafkaClientAdapter).GetMethod( @@ -343,4 +427,3 @@ private static MethodInfo GetDekafKafkaMethod(string name) } } } -#endif diff --git a/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaIntegrationDockerTests.cs b/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaIntegrationDockerTests.cs index c2a24bd87d..beca9fb7f3 100644 --- a/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaIntegrationDockerTests.cs +++ b/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaIntegrationDockerTests.cs @@ -58,10 +58,9 @@ public async Task OneTimeSetUpAsync() { try { - // Dekaf (the net10.0 client) requires a Kafka 4.0+ broker for the - // KIP-848 ConsumerGroupHeartbeat API; the Apache image runs KRaft and - // enables the new consumer group protocol by default. The Confluent - // client used on other TFMs falls back to the classic protocol. + // Dekaf requires a Kafka 4.0+ broker for the KIP-848 + // ConsumerGroupHeartbeat API; the Apache image runs KRaft and enables + // the new consumer group protocol by default. m_container = new KafkaBuilder("apache/kafka:4.0.0").Build(); await m_container.StartAsync().ConfigureAwait(false); } @@ -128,8 +127,8 @@ public async Task RealBrokerRoundTripsPayloadAsync() byte[] payload = [0x10, 0x20, 0x30, 0x40]; // Open the publisher and produce one record before the subscriber - // subscribes so the topic already exists: the Dekaf consumer (net10, - // KIP-848 protocol) does not pick up a topic that is created after it + // subscribes so the topic already exists: the Dekaf consumer (KIP-848 + // protocol) does not pick up a topic that is created after it // has subscribed. The subscriber reads from the earliest offset once // it is assigned the partition. await publisher.OpenAsync(CancellationToken.None).ConfigureAwait(false); diff --git a/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaTransportServiceCollectionExtensionsTests.cs b/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaTransportServiceCollectionExtensionsTests.cs index 96499b3337..b280b1bfb4 100644 --- a/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaTransportServiceCollectionExtensionsTests.cs +++ b/tests/Opc.Ua.PubSub.Kafka.Tests/KafkaTransportServiceCollectionExtensionsTests.cs @@ -34,6 +34,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; +using Moq; using NUnit.Framework; using Opc.Ua.PubSub.Application; using Opc.Ua.PubSub.Kafka.Internal; @@ -84,7 +85,40 @@ public async Task AddKafkaTransportWithCallbackRegistersFactoriesAndOptionsAsync KafkaProfiles.PubSubKafkaJsonTransport, KafkaProfiles.PubSubKafkaUadpTransport ])); - Assert.That(serviceProvider.GetRequiredService(), Is.Not.Null); + Assert.That( + serviceProvider.GetRequiredService(), + Is.InstanceOf()); + } + + [Test] + public void WithConfluentKafkaClientReplacesDefaultFactory() + { + var services = new ServiceCollection(); + services.AddSingleton(); + var builder = new Mock(MockBehavior.Strict); + builder.SetupGet(value => value.Services).Returns(services); + + IPubSubBuilder result = builder.Object.WithConfluentKafkaClient(); + ServiceDescriptor descriptor = services.Single( + service => service.ServiceType == typeof(IKafkaClientFactory)); + + Assert.Multiple(() => + { + Assert.That(result, Is.SameAs(builder.Object)); + Assert.That(descriptor.ImplementationType, Is.EqualTo(typeof(ConfluentKafkaClientFactory))); + Assert.That(descriptor.Lifetime, Is.EqualTo(ServiceLifetime.Singleton)); + }); + builder.VerifyGet(value => value.Services, Times.Once); + } + + [Test] + public void WithConfluentKafkaClientWithNullBuilderThrows() + { + IPubSubBuilder builder = null!; + + Assert.That( + () => builder.WithConfluentKafkaClient(), + Throws.TypeOf()); } [Test]