MASES.KNet 1.4.2

.NET 6.0 .NET Core 3.1 .NET Framework 4.6.2
There is a newer version of this package available.
See the version list below for details.
dotnet add package MASES.KNet --version 1.4.2
NuGet\Install-Package MASES.KNet -Version 1.4.2
This command is intended to be used within the Package Manager Console in Visual Studio, as it uses the NuGet module's version of Install-Package.
<PackageReference Include="MASES.KNet" Version="1.4.2" />
For projects that support PackageReference, copy this XML node into the project file to reference the package.
paket add MASES.KNet --version 1.4.2
#r "nuget: MASES.KNet, 1.4.2"
#r directive can be used in F# Interactive, C# scripting and .NET Interactive. Copy this into the interactive tool or source code of the script to reference the package.
// Install MASES.KNet as a Cake Addin
#addin nuget:?package=MASES.KNet&version=1.4.2

// Install MASES.KNet as a Cake Tool
#tool nuget:?package=MASES.KNet&version=1.4.2

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

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 MASES.KNet.Clients.Producer;
using Java.Util;
using System;

namespace MASES.KNetTemplate.KNetProducer
{
    class Program
    {
        const string theServer = "localhost:9092";
        const string theTopic = "myTopic";

        static string serverToUse = theServer;
        static string topicToUse = theTopic;

        static void Main(string[] args)
        {
            var appArgs = KNetCore.ApplicationArgs;

            if (appArgs.Length != 0)
            {
                serverToUse = args[0];
            }

			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");

			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++;
				}
			}
        }
    }
}

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 MASES.KNet.Clients.Producer;
using Java.Util;
using System;

namespace MASES.KNetTemplate.KNetProducer
{
    class Program
    {
        const string theServer = "localhost:9092";
        const string theTopic = "myTopic";

        static string serverToUse = theServer;
        static string topicToUse = theTopic;

        static void Main(string[] args)
        {
            var appArgs = KNetCore.ApplicationArgs;

            if (appArgs.Length != 0)
            {
                serverToUse = args[0];
            }

			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");

			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++;
					}
				}
			}
        }
    }
}

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 MASES.KNet.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 void Main(string[] args)
        {
            var appArgs = KNetCore.ApplicationArgs;

            if (appArgs.Length != 0)
            {
                serverToUse = args[0];
            }

            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");

            using (var consumer = new KafkaConsumer<string, string>(props))
            {
                consumer.Subscribe(Collections.singleton(topicToUse));
                while (true)
                {
                    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}");
                    }
                }
            }
        }
    }
}

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
.NET net5.0 net5.0-windows net6.0 net6.0-android net6.0-ios net6.0-maccatalyst net6.0-macos net6.0-tvos net6.0-windows net7.0 net7.0-android net7.0-ios net7.0-maccatalyst net7.0-macos net7.0-tvos net7.0-windows
.NET Core netcoreapp3.1
.NET Framework net462 net463 net47 net471 net472 net48 net481
Compatible target framework(s)
Additional computed target framework(s)
Learn more about Target Frameworks and .NET Standard.

NuGet packages (1)

Showing the top 1 NuGet packages that depend on MASES.KNet:

Package Downloads
MASES.EntityFrameworkCore.KNet

EntityFrameworkCore KNet - EntityFrameworkCore provider for Apache Kafka

GitHub repositories

This package is not used by any popular GitHub repositories.

Version Downloads Last updated
1.4.8 142,427 11/28/2022
1.4.7 176 11/23/2022
1.4.6 166 11/22/2022
1.4.5 166 11/21/2022
1.4.4 18,649 11/1/2022
1.4.3 18,619 10/21/2022
1.4.2 223 10/17/2022
1.4.1 325 10/10/2022
1.4.0 51,442 10/6/2022
1.3.6 35,552 9/19/2022
1.3.5 53,027 9/8/2022
1.3.4 65,138 8/18/2022
1.3.3 375 8/5/2022
1.3.2 392 6/19/2022
1.3.1 34,307 5/23/2022
1.2.4 469 5/11/2022
1.2.3 415 5/7/2022
1.2.2 420 5/2/2022
1.2.1 4,943 3/28/2022
1.2.0 421 3/20/2022