MASES.KNet
2.8.1
dotnet add package MASES.KNet --version 2.8.1
NuGet\Install-Package MASES.KNet -Version 2.8.1
<PackageReference Include="MASES.KNet" Version="2.8.1" />
paket add MASES.KNet --version 2.8.1
#r "nuget: MASES.KNet, 2.8.1"
// Install MASES.KNet as a Cake Addin #addin nuget:?package=MASES.KNet&version=2.8.1 // Install MASES.KNet as a Cake Tool #tool nuget:?package=MASES.KNet&version=2.8.1
title: Usage of .NET suite for Apache Kafka™ _description: Describes how to use .NET suite for Apache Kafka™
KNet: library usage
To use KNet classes the developer can write code in .NET using the same classes available in the official Apache Kafka™ package. If classes or methods are not available yet it is possible to use the approach synthetized in What to do if an API was not yet implemented
Environment setup
KNet accepts many command-line switches to customize its behavior. The full list is available at Command line switch page.
JVM identification
One of the most important command-line switch is JVMPath and it is available in JCOBridge switches: it can be used to set-up the location of the JVM library if JCOBridge is not able to identify a suitable JRE/JDK installation. If a developer is using KNet within its own product it is possible to override the JVMPath property with a snippet like the following one:
class MyKNetCore : KNetCore
{
public override string JVMPath
{
get
{
string pathToJVM = "Set here the path to JVM library or use your own search method";
return pathToJVM;
}
}
}
IMPORTANT NOTE: pathToJVM
shall be escaped
string pathToJVM = "C:\\Program Files\\Eclipse Adoptium\\jdk-11.0.18.10-hotspot\\bin\\server\\jvm.dll";
string pathToJVM = @"C:\Program Files\Eclipse Adoptium\jdk-11.0.18.10-hotspot\bin\server\jvm.dll";
Special initialization conditions
JCOBridge try to identify a suitable JRE/JDK installation within the system using some standard mechanism of JRE/JDK: JAVA_HOME
environment variable or Windows registry if available.
However it is possible, on Windows operating systems, that the library raises an InvalidOperationException: Missing Java Key in registry: Couldn't find Java installed on the machine.
This means that neither JAVA_HOME
nor Windows registry contains information about a default installed JRE/JDK: some vendors may not setup them.
If the developer/user encounter this condition can do the following steps:
- On a command prompt execute
set | findstr JAVA_HOME
and verify the result; - If something was reported maybe the
JAVA_HOME
environment variable is not set at system level, but at a different level like user level which is not visible from the KNet process that raised the exception; - Try to set
JAVA_HOME
at system level e.g.JAVA_HOME=C:\Program Files\Eclipse Adoptium\jdk-11.0.18.10-hotspot\
; - Try to set
JCOBRIDGE_JVMPath
at system level e.g.JCOBRIDGE_JVMPath=C:\Program Files\Eclipse Adoptium\jdk-11.0.18.10-hotspot\
.
IMPORTANT NOTES:
- One of
JCOBRIDGE_JVMPath
orJAVA_HOME
environment variables or Windows registry (on Windows OSes) shall be available JCOBRIDGE_JVMPath
environment variable takes precedence overJAVA_HOME
and Windows registry: you can setJCOBRIDGE_JVMPath
toC:\Program Files\Eclipse Adoptium\jdk-11.0.18.10-hotspot\bin\server\jvm.dll
and avoid to overrideJVMPath
in your code- After first initialization steps,
JVMPath
takes precedence overJCOBRIDGE_JVMPath
/JAVA_HOME
environment variables or Windows registry
Producer example
Below the reader can found two different version of producer examples.
Simple producer
A basic producer can be like the following one:
using MASES.KNet;
using Org.Apache.Kafka™.Clients.Producer;
using Java.Util;
using System;
using System.Threading;
namespace MASES.KNetTemplate.KNetProducer
{
class Program
{
const string theServer = "localhost:9092";
const string theTopic = "myTopic";
static string serverToUse = theServer;
static string topicToUse = theTopic;
static readonly ManualResetEvent resetEvent = new ManualResetEvent(false);
static void Main(string[] args)
{
KNetCore.CreateGlobalInstance();
var appArgs = KNetCore.FilteredArgs;
if (appArgs.Length != 0)
{
serverToUse = args[0];
}
/**** Direct mode ******
Properties props = new Properties();
props.Put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverToUse);
props.Put(ProducerConfig.ACKS_CONFIG, "all");
props.Put(ProducerConfig.RETRIES_CONFIG, 0);
props.Put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.Put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.Put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
******/
Properties props = ProducerConfigBuilder.Create()
.WithBootstrapServers(serverToUse)
.WithAcks(ProducerConfig.Acks.All)
.WithRetries(0)
.WithLingerMs(1)
.WithKeySerializerClass("org.apache.kafka.common.serialization.StringSerializer")
.WithValueSerializerClass("org.apache.kafka.common.serialization.StringSerializer")
.ToProperties();
Console.CancelKeyPress += Console_CancelKeyPress;
Console.WriteLine("Press Ctrl-C to exit");
using (KafkaProducer producer = new KafkaProducer(props))
{
int i = 0;
while (!resetEvent.WaitOne(0))
{
var record = new ProducerRecord<string, string>(topicToUse, i.ToString(), i.ToString());
var result = producer.Send(record);
Console.WriteLine($"Producing: {record} with result: {result.Get()}");
producer.Flush();
i++;
}
}
}
private static void Console_CancelKeyPress(object sender, ConsoleCancelEventArgs e)
{
if (e.Cancel) resetEvent.Set();
}
}
}
The example above can be found in the templates package. Its behavior is:
- during initialization prepares the properties,
- create a producer using the properties
- create ProducerRecord and send it
- print out the produced data and the resulting RecordMetadata
Producer with Callback
A producer with Callback can be like the following one. In this example the reader can highlight a slightly difference from the corresponding Java code. Surf JVM callbacks to go into detail in the callback management from JVM.
using MASES.KNet;
using Org.Apache.Kafka™.Clients.Producer;
using Java.Util;
using System;
using System.Threading;
namespace MASES.KNetTemplate.KNetProducer
{
class Program
{
const string theServer = "localhost:9092";
const string theTopic = "myTopic";
static string serverToUse = theServer;
static string topicToUse = theTopic;
static readonly ManualResetEvent resetEvent = new ManualResetEvent(false);
static void Main(string[] args)
{
KNetCore.CreateGlobalInstance();
var appArgs = KNetCore.FilteredArgs;
if (appArgs.Length != 0)
{
serverToUse = args[0];
}
/**** Direct mode ******
Properties props = new Properties();
props.Put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverToUse);
props.Put(ProducerConfig.ACKS_CONFIG, "all");
props.Put(ProducerConfig.RETRIES_CONFIG, 0);
props.Put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.Put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.Put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
******/
Properties props = ProducerConfigBuilder.Create()
.WithBootstrapServers(serverToUse)
.WithAcks(ProducerConfig.Acks.All)
.WithRetries(0)
.WithLingerMs(1)
.WithKeySerializerClass("org.apache.kafka.common.serialization.StringSerializer")
.WithValueSerializerClass("org.apache.kafka.common.serialization.StringSerializer")
.ToProperties();
Console.CancelKeyPress += Console_CancelKeyPress;
Console.WriteLine("Press Ctrl-C to exit");
using (KafkaProducer producer = new KafkaProducer(props))
{
int i = 0;
using (var callback = new Callback((o1, o2) =>
{
if (o2 != null) Console.WriteLine(o2.ToString());
else Console.WriteLine($"Produced on topic {o1.Topic} at offset {o1.Offset}");
}))
{
while (!resetEvent.WaitOne(0))
{
var record = new ProducerRecord<string, string>(topicToUse, i.ToString(), i.ToString());
var result = producer.Send(record, callback);
Console.WriteLine($"Producing: {record} with result: {result.Get()}");
producer.Flush();
i++;
}
}
}
}
private static void Console_CancelKeyPress(object sender, ConsoleCancelEventArgs e)
{
if (e.Cancel) resetEvent.Set();
}
}
}
The example above can be found in the templates package. Its behavior is:
- during initialization prepares the properties
- create a producer using the properties
- create ProducerRecord and send it using the API Send with the attached Callback
- when the operation completed the Callback is called:
- if an Exception was raised it will be printed out
- otherwise the RecordMetadata is printed out
- print out the produced data and the resulting RecordMetadata
Consumer example
A basic consumer can be like the following one:
using MASES.KNet;
using Org.Apache.Kafka™.Clients.Consumer;
using Java.Util;
using System;
namespace MASES.KNetTemplate.KNetConsumer
{
class Program
{
const string theServer = "localhost:9092";
const string theTopic = "myTopic";
static string serverToUse = theServer;
static string topicToUse = theTopic;
static readonly ManualResetEvent resetEvent = new ManualResetEvent(false);
static void Main(string[] args)
{
KNetCore.CreateGlobalInstance();
var appArgs = KNetCore.FilteredArgs;
if (appArgs.Length != 0)
{
serverToUse = args[0];
}
/**** Direct mode ******
Properties props = new Properties();
props.Put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serverToUse);
props.Put(ConsumerConfig.GROUP_ID_CONFIG, "test");
props.Put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.Put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.Put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.Put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
*******/
Properties props = ConsumerConfigBuilder.Create()
.WithBootstrapServers(serverToUse)
.WithGroupId("test")
.WithEnableAutoCommit(true)
.WithAutoCommitIntervalMs(1000)
.WithKeyDeserializerClass("org.apache.kafka.common.serialization.StringDeserializer")
.WithValueDeserializerClass("org.apache.kafka.common.serialization.StringDeserializer")
.ToProperties();
Console.CancelKeyPress += Console_CancelKeyPress;
Console.WriteLine("Press Ctrl-C to exit");
using (var consumer = new KafkaConsumer<string, string>(props))
{
var topics = Collections.Singleton(topicToUse);
consumer.Subscribe(topics);
while (!resetEvent.WaitOne(0))
{
var records = consumer.Poll((long)TimeSpan.FromMilliseconds(200).TotalMilliseconds);
foreach (var item in records)
{
Console.WriteLine($"Offset = {item.Offset}, Key = {item.Key}, Value = {item.Value}");
}
}
topics?.Dispose(); // needed to avoid Java.Lang.NullPointerException in some conditions where .NET GC retires topics too early
}
}
private static void Console_CancelKeyPress(object sender, ConsoleCancelEventArgs e)
{
if (e.Cancel) resetEvent.Set();
}
}
}
The example above can be found in the templates package. Its behavior is:
- during initialization prepares the properties,
- create a consumer using the properties
- subscribe and starts consume
- when data are received it logs to the console the information.
Product | Versions Compatible and additional computed target framework versions. |
---|---|
.NET | net6.0 is compatible. net6.0-android was computed. net6.0-ios was computed. net6.0-maccatalyst was computed. net6.0-macos was computed. net6.0-tvos was computed. net6.0-windows was computed. net7.0 was computed. net7.0-android was computed. net7.0-ios was computed. net7.0-maccatalyst was computed. net7.0-macos was computed. net7.0-tvos was computed. net7.0-windows was computed. net8.0 is compatible. net8.0-android was computed. net8.0-browser was computed. net8.0-ios was computed. net8.0-maccatalyst was computed. net8.0-macos was computed. net8.0-tvos was computed. net8.0-windows was computed. |
.NET Framework | net462 is compatible. net463 was computed. net47 was computed. net471 was computed. net472 was computed. net48 was computed. net481 was computed. |
-
.NETFramework 4.6.2
- MASES.JNet (>= 2.5.9)
-
net6.0
- MASES.JNet (>= 2.5.9)
-
net8.0
- MASES.JNet (>= 2.5.9)
NuGet packages (5)
Showing the top 5 NuGet packages that depend on MASES.KNet:
Package | Downloads |
---|---|
MASES.EntityFrameworkCore.KNet.Serialization
EntityFrameworkCore KNet - Serialization support for EntityFrameworkCore provider for Apache Kafka |
|
MASES.KNet.Serialization.Avro
Avro Serializer/Deserializer of .NET suite for Apache Kafka. KNet is a comprehensive .NET suite for Apache Kafka providing all features: Producer, Consumer, Admin, Streams, Connect, backends (ZooKeeper and Kafka). |
|
MASES.KNet.Serialization.Json
Json Serializer/Deserializer of .NET suite for Apache Kafka. KNet is a comprehensive .NET suite for Apache Kafka providing all features: Producer, Consumer, Admin, Streams, Connect, backends (ZooKeeper and Kafka). |
|
MASES.KNet.Serialization.MessagePack
MessagePack Serializer/Deserializer of .NET suite for Apache Kafka. KNet is a comprehensive .NET suite for Apache Kafka providing all features: Producer, Consumer, Admin, Streams, Connect, backends (ZooKeeper and Kafka). |
|
MASES.KNet.Serialization.Protobuf
Protobuf Serializer/Deserializer of .NET suite for Apache Kafka. KNet is a comprehensive .NET suite for Apache Kafka providing all features: Producer, Consumer, Admin, Streams, Connect, backends (ZooKeeper and Kafka). |
GitHub repositories
This package is not used by any popular GitHub repositories.
Version | Downloads | Last updated |
---|---|---|
2.8.1 | 74,578 | 9/20/2024 |
2.8.0 | 41,330 | 8/6/2024 |
2.7.9 | 468 | 9/20/2024 |
2.7.8 | 13,860 | 7/31/2024 |
2.7.7 | 5,776 | 7/30/2024 |
2.7.6 | 268 | 7/29/2024 |
2.7.5 | 32,258 | 7/2/2024 |
2.7.4 | 55,430 | 6/27/2024 |
2.7.3 | 28,163 | 6/24/2024 |
2.7.2 | 41,903 | 5/25/2024 |
2.7.1 | 13,946 | 5/18/2024 |
2.7.0 | 803 | 5/16/2024 |
2.6.6 | 376 | 9/20/2024 |
2.6.5 | 435 | 9/16/2024 |
2.6.4 | 418 | 6/14/2024 |
2.6.3 | 334 | 6/11/2024 |
2.6.2 | 619 | 5/17/2024 |
2.6.1 | 18,408 | 5/3/2024 |
2.6.0 | 139,662 | 3/1/2024 |
2.5.0 | 654 | 2/28/2024 |
2.4.3 | 58,306 | 2/11/2024 |
2.4.2 | 47,372 | 1/27/2024 |
2.4.1 | 11,670 | 1/21/2024 |
2.4.0 | 351 | 1/20/2024 |
2.3.0 | 92,998 | 11/25/2023 |
2.2.0 | 48,491 | 10/19/2023 |
2.1.3 | 26,509 | 10/11/2023 |
2.1.2 | 26,721 | 10/6/2023 |
2.1.1 | 1,407 | 10/5/2023 |
2.1.0 | 975 | 9/27/2023 |
2.0.2 | 135,925 | 8/2/2023 |
2.0.1 | 10,049 | 7/11/2023 |
2.0.0 | 37,405 | 7/8/2023 |
1.5.5 | 32,628 | 7/1/2023 |
1.5.4 | 25,567 | 5/28/2023 |
1.5.3 | 46,239 | 4/16/2023 |
1.5.2 | 33,281 | 4/11/2023 |
1.5.1 | 65,156 | 3/15/2023 |
1.5.0 | 55,773 | 2/9/2023 |
1.4.8 | 143,667 | 11/28/2022 |
1.4.7 | 783 | 11/23/2022 |
1.4.6 | 868 | 11/22/2022 |
1.4.5 | 724 | 11/21/2022 |
1.4.4 | 19,786 | 11/1/2022 |
1.4.3 | 19,966 | 10/21/2022 |
1.4.2 | 877 | 10/17/2022 |
1.4.1 | 1,454 | 10/10/2022 |
1.4.0 | 52,692 | 10/6/2022 |
1.3.6 | 36,926 | 9/19/2022 |
1.3.5 | 54,257 | 9/8/2022 |
1.3.4 | 66,301 | 8/18/2022 |
1.3.3 | 916 | 8/5/2022 |
1.3.2 | 912 | 6/19/2022 |
1.3.1 | 35,611 | 5/23/2022 |
1.2.4 | 1,057 | 5/11/2022 |
1.2.3 | 971 | 5/7/2022 |
1.2.2 | 969 | 5/2/2022 |
1.2.1 | 6,040 | 3/28/2022 |
1.2.0 | 1,547 | 3/20/2022 |