Skip to content

Use Dekaf by default for Kafka on all target frameworks #4038

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
marcschier wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Use Dekaf by default for Kafka on all target frameworks #4038

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
554c86e
Use Dekaf for Kafka on all target frameworks
marcschier Jul 19, 2026
72164b8
Address Kafka PR review feedback
marcschier Jul 19, 2026
553c62d
Restore Confluent Kafka opt-in backend
marcschier Jul 20, 2026
44ab552
Add Kafka patch coverage tests
marcschier Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 10 additions & 7 deletions Directory.Packages.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,10 @@
<PackageVersion Include="BenchmarkDotNet" Version="0.15.8" />
<PackageVersion Include="BitFaster.Caching" Version="2.6.0" />
<PackageVersion Include="BouncyCastle.Cryptography" Version="2.6.2" />
<PackageVersion Include="Confluent.Kafka" Version="2.15.0" />
Comment thread
marcschier marked this conversation as resolved.
<PackageVersion Include="coverlet.collector" Version="10.0.1" />
<PackageVersion Include="Crdt" Version="1.1.0" />
<PackageVersion Include="Crdt.Transport" Version="1.1.0" />
<PackageVersion Include="Dekaf" Version="1.2.0" />
<PackageVersion Include="Dekaf" Version="1.7.0" />
<PackageVersion Include="DotNext" Version="5.26.3" />
<PackageVersion Include="RaftCs" Version="1.1.0" />
<PackageVersion Include="RaftCs.Transport" Version="1.1.0" />
Expand All @@ -38,7 +37,11 @@
<PackageVersion Include="Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets" Version="2.3.11" />
<PackageVersion Include="Microsoft.AspNetCore.WebSockets" Version="2.3.11" />
<PackageVersion Include="Microsoft.Bcl.HashCode" Version="6.0.0" />
<PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.9" />
<!-- Dekaf 1.7.0 requires AsyncInterfaces, Logging.Abstractions and Options
Comment thread
marcschier marked this conversation as resolved.
Outdated
10.0.10. Their legacy-target dependency closure also requires
DependencyInjection.Abstractions and DiagnosticSource 10.0.10. Other
10.0.9 pins are outside that closure and intentionally unchanged. -->
<PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="10.0.10" />
Comment thread
marcschier marked this conversation as resolved.
<PackageVersion Include="Microsoft.Bcl.TimeProvider" Version="10.0.9" />
<PackageVersion Include="Microsoft.CodeAnalysis.Analyzers" Version="4.14.0" />
<PackageVersion Include="Microsoft.CodeAnalysis.Common" Version="4.14.0" />
Expand All @@ -49,16 +52,16 @@
<PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Logging" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.10" />
<PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.10" />
Comment thread
marcschier marked this conversation as resolved.
<PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Diagnostics" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Http" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Http.Resilience" Version="10.7.0" />
<PackageVersion Include="Microsoft.Extensions.Hosting" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Options" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.Options" Version="10.0.10" />
Comment thread
marcschier marked this conversation as resolved.
<PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.9" />
<PackageVersion Include="Microsoft.Extensions.TimeProvider.Testing" Version="10.7.0" />
<PackageVersion Include="ModelContextProtocol" Version="1.4.0" />
Expand Down Expand Up @@ -108,7 +111,7 @@
<PackageVersion Include="System.Buffers" Version="4.6.1" />
<PackageVersion Include="System.Composition.AttributedModel" Version="10.0.9" />
<PackageVersion Include="System.Runtime.CompilerServices.Unsafe" Version="6.1.2" />
<PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="10.0.9" />
<PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="10.0.10" />
Comment thread
marcschier marked this conversation as resolved.
<PackageVersion Include="System.Memory" Version="4.6.3" />
<PackageVersion Include="System.Formats.Asn1" Version="10.0.9" />
<PackageVersion Include="System.IO.Compression" Version="4.3.0" />
Expand Down
10 changes: 3 additions & 7 deletions docs/PubSub.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 │
└─────────────────┘ └──────────────────────┘ └────────────────────┘
```

Expand Down Expand Up @@ -764,7 +764,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 transport uses the pure-managed [Dekaf](https://github.com/thomhurst/Dekaf) client on every supported target framework, with no native librdkafka dependency. This repository asserts and validates NativeAOT/trimming compatibility on `net10.0`; the other target frameworks use the same managed client but are not part of the repository's AOT validation matrix. See [Native AOT](#native-aot).

## Encodings

Expand Down Expand Up @@ -1826,11 +1826,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`). The Kafka transport uses the managed Dekaf client on every target framework and has no native librdkafka dependency (see [Apache Kafka](#apache-kafka)).

- **No reflection-based serialization.** Source-generated
`IEncodeable` types (Part 14 datatypes) plus hand-written
Expand Down
4 changes: 0 additions & 4 deletions src/Opc.Ua.PubSub.Kafka/DependencyInjection/KafkaTransportServiceCollectionExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,11 +187,7 @@ public static IOpcUaBuilder AddKafkaPubSub(

private static void RegisterShared(IServiceCollection services)
{
#if NET10_0_OR_GREATER
services.TryAddSingleton<IKafkaClientFactory, DekafKafkaClientFactory>();
#else
services.TryAddSingleton<IKafkaClientFactory, ConfluentKafkaClientFactory>();
#endif
services.AddPubSubTransportFactory(sp =>
new KafkaPubSubTransportFactory(
KafkaProfiles.PubSubKafkaJsonTransport,
Expand Down
1 change: 0 additions & 1 deletion src/Opc.Ua.PubSub.Kafka/EventIds.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,6 @@ namespace Opc.Ua
/// </remarks>
internal static class PubSubKafkaEventIds
{
public const int ConfluentKafkaClientAdapter = 0;
public const int KafkaBrokerTransport = 10;
public const int KafkaClientAdapter = 20;
}
Expand Down
Loading
Loading