Confluent .NET Client for Apache Kafka

In the KIP-848 protocol, rebalancing is fully incremental.

Critical Change: The partitions list passed to PartitionsAssignedHandler() and PartitionsRevokedHandler() no longer contains the full assignment. Instead, it contains only the incremental changes (the specific partitions being added or the specific partitions being revoked).

Ensure your logic handles these partial lists rather than assuming they represent the complete state of the consumer's assignment. Additionally, all assignors under KIP-848 (including range) are now sticky.

This example demonstrates how to implement exactly once processing semantics for a streaming map-reduce calculation (specifically a WordCount application) on older Kafka environments (Apache Kafka <= 2.4.0 and Confluent.Kafka <= 1.6.0).

Note: For modern Kafka versions, refer to KIP-447, which provides a significantly easier way to achieve exactly once semantics.

The implementation follows a two-step process:

1. MAP: Reads from the lines topic, splits text into words, and writes to a words topic (partitioned by word).
2. REDUCE: Reads from the words topic and maintains a running tally. State is stored permanently in a counts topic and cached locally in RocksDb for aggregation and ad-hoc lookups.

The Confluent.SchemaRegistry.Encryption project uses vendored (copied) versions of Miscreant and HKDF.Standard instead of standard NuGet packages. This is done to ensure all assemblies are strongly named, maintaining assembly signing compatibility required by the project.

Because these are vendored, the original namespaces have been changed to prevent conflicts and ensure they are part of the Confluent assembly hierarchy:

• Miscreant is now under the namespace: Confluent.SchemaRegistry.Encryption.Vendored.Miscreant
• HKDF.Standard is now under the namespace: Confluent.SchemaRegistry.Encryption.Vendored.HkdfStandard

When using GroupInstanceId for static group membership in KIP-848, the fencing behavior has changed:

• Fencing Target: A newly joining member is fenced with UnreleasedInstanceId (fatal) if a duplicate GroupInstanceId is detected. (In the classic protocol, the existing member was fenced instead).
• Best Practices:
  • Ensure only one active instance exists per GroupInstanceId.
  • Consumers must shut down cleanly to avoid blocking replacements until the session timeout expires.

The vendored HKDF.Standard library provides an HMAC-based Extract-and-Expand Key Derivation Function (HKDF) that is RFC 5869 compliant. It is used within LocalKmsClient.cs for deriving keys from secrets.

Key details for consumers:

• Namespace: Confluent.SchemaRegistry.Encryption.Vendored.HkdfStandard
• Implementation: A simplified version that focuses on the DeriveKey() method to ensure compatibility across all supported target frameworks using standard byte[] arrays.

The vendored Miscreant library provides AES-SIV (Synthetic Initialization Vector) encryption and decryption. In the Confluent.SchemaRegistry.Encryption project, it is utilized by Cryptor.cs and KmsClients.cs for core cryptographic operations.

When consuming these features, note that the library has been re-namespaced to Confluent.SchemaRegistry.Encryption.Vendored.Miscreant.

For production applications, it is recommended to serialize data using the Confluent Schema Registry to ensure data quality and governance. The client provides 'Serdes' (Serializer/Deserializer) packages for:

• JSON: High support in .NET.
• Protobuf: High support in .NET.
• Avro: Well-suited for streaming, though non-Java implementation maturity may vary.

To use them, you must provide a CachedSchemaRegistryClient to your ProducerBuilder or ConsumerBuilder via .SetValueSerializer() or .SetValueDeserializer().

To add AWS IAM-based authentication to your Kafka clients, add the Confluent.Kafka.OAuthBearer.Aws package. This package is opt-in and config-activated; adding the package and setting the required configuration keys is sufficient for most use cases without requiring code changes at the integration site.

Note: This package versions in lockstep with Confluent.Kafka. Always ensure you are referencing matching versions.

<ItemGroup>
  <PackageReference Include="Confluent.Kafka" Version="2.15.1" />
  <PackageReference Include="Confluent.Kafka.OAuthBearer.Aws" Version="2.15.1" />
</ItemGroup>

Once a Kafka producer service (like KafkaDependentProducer) is registered in the Dependency Injection (DI) container, it can be injected into various parts of the ASP.NET Core pipeline:

• Controllers: Inject the producer service into a controller (e.g., HomeController) to produce messages as part of standard web requests.
• Middleware: Inject the producer service into middleware (e.g., RequestTimerMiddleware) to perform cross-cutting concerns, such as logging request metrics or timing information to Kafka for every incoming request.