dannyb2018 / kafka-streams-dotnet

.NET Stream Processing Library for Apache Kafka 🚀 /*** WORK IN PROGRESS ***/

.NET Stream Processing Library for Apache Kafka ^TM ·

Sonarcloud statistics

GitHub statistics

---

Streamiz Kafka .NET is .NET stream processing library for Apache Kafka.

It's allowed to develop .NET applications that transform input Kafka topics into output Kafka topics. It's supported .NET Standard 2.1.

It's a rewriting inspired by Kafka Streams.

Finally it will provide the same functionality as Kafka Streams.

At moment, this project is being written. Thanks for you contribution !

Usage

Sample code

static void Main(string[] args)
{
    CancellationTokenSource source = new CancellationTokenSource();
    
    var config = new StreamConfig<StringSerDes, StringSerDes>();
    config.ApplicationId = "test-app";
    config.BootstrapServers = "192.168.56.1:9092";
    config.SaslMechanism = SaslMechanism.Plain;
    config.SaslUsername = "admin";
    config.SaslPassword = "admin";
    config.SecurityProtocol = SecurityProtocol.SaslPlaintext;
    config.AutoOffsetReset = AutoOffsetReset.Earliest;
    config.NumStreamThreads = 2;
    
    StreamBuilder builder = new StreamBuilder();

    builder.Stream<string, string>("test")
        .FilterNot((k, v) => v.Contains("test"))
        .Peek((k,v) => Console.WriteLine($"Key : {k} | Value : {v}"))
        .To("test-output");

    builder.Table("topic", InMemory<string, string>.As("test-ktable-store"));

    Topology t = builder.Build();
    KafkaStream stream = new KafkaStream(t, config);

    Console.CancelKeyPress += (o, e) => {
        source.Cancel();
        stream.Close();
    };

    stream.Start(source.Token);
}

Timeline

• End May 2020 - Beta 0.0.1 - All stateless processors, Exactly Once Semantic, InMemory store
• End October 2020 - Beta 0.0.2 - All statefull processors, Global Store, RocksDB Store
• End 2020 / Begin 2021 - 1.0.0 RC1 - Processor API, Metrics, Interactive Queries

Stateless processor implemention

Operator Name	Method	TODO	IMPLEMENTED	TESTED	DOCUMENTED
Branch	KStream -> KStream[]				☑
Filter	KStream -> KStream				☑
Filter	KTable -> KTable				☑
InverseFilter	KStream -> KStream				☑
InverseFilter	KTable -> KTable				☑
FlatMap	KStream → KStream				☑
FlatMapValues	KStream → KStream				☑
Foreach	KStream → void				☑
GroupByKey	KStream → KGroupedStream				☑
GroupBy	KStream → KGroupedStream				☑
GroupBy	KTable → KGroupedTable				☑
Map	KStream → KStream				☑
MapValues	KStream → KStream				☑
MapValues	KTable → KTable				☑
Peek	KStream → KStream				☑
Print	KStream → void				☑
SelectKey	KStream → KStream				☑
Table to Steam	KTable → KStream				☑

Statefull processor implementation

Operator Name	Method	TODO	IMPLEMENTED	TESTED	DOCUMENTED
Aggregate	KGroupedStream -> KTable	☑
Aggregate	KGroupedTable -> KTable	☑
Aggregate(windowed)	KGroupedStream -> KTable	☑
Count	KGroupedStream -> KTable	☑
Count	KGroupedTable -> KTable	☑
Count(windowed)	KGroupedStream → KStream	☑
Reduce	KGroupedStream → KTable	☑
Reduce	KGroupedTable → KTable	☑
Reduce(windowed)	KGroupedStream → KTable	☑
InnerJoin(windowed)	(KStream,KStream) → KStream	☑
LeftJoin(windowed)	(KStream,KStream) → KStream	☑
OuterJoin(windowed)	(KStream,KStream) → KStream	☑
InnerJoin(windowed)	(KTable,KTable) → KTable	☑
LeftJoin(windowed)	(KTable,KTable) → KTable	☑
OuterJoin(windowed)	(KTable,KTable) → KTable	☑
InnerJoin(windowed)	(KStream,KTable) → KStream	☑
LeftJoin(windowed)	(KStream,KTable) → KStream	☑
InnerJoin(windowed)	(KStream,GlobalKTable) → KStream	☑
LeftJoin(windowed)	(KStream,GlobalKTable) → KStream	☑

Test topology driver

Must be used for testing your stream topology. Simulate a kafka cluster in memory. Usage:

static void Main(string[] args)
{
    var config = new StreamConfig<StringSerDes, StringSerDes>();
    config.ApplicationId = "test-test-driver-app";
    
    StreamBuilder builder = new StreamBuilder();

    builder.Stream<string, string>("test")
        .Filter((k, v) => v.Contains("test"))
        .To("test-output");

    Topology t = builder.Build();

    using (var driver = new TopologyTestDriver(t, config))
    {
        var inputTopic = driver.CreateInputTopic<string, string>("test");
        var outputTopic = driver.CreateOuputTopic<string, string>("test-output", TimeSpan.FromSeconds(5));
        inputTopic.PipeInput("test", "test-1234");
        var r = outputTopic.ReadKeyValue();
    }
}

TODO implementation

• Statefull processors impl [ ]
• Subtopology impl [ ]
• Task restoring [ ]
• Topology description [ ]
• Global state store [ ]
• Processor API [ ] + Refactor topology node processor builder [ ]
• Repartition impl [ ]
• Unit tests (TestTopologyDriver, ...) [ ]
• Rocks DB state implementation [ ]
• Optimizing Kafka Streams Topologies [ ]
• Interactive Queries [ ]
• Metrics [ ]

Some documentations for help during implementation : https://docs.confluent.io/current/streams/index.html

https://kafka.apache.org/20/documentation/streams/developer-guide/dsl-api.html#stateless-transformations