diff --git a/src/Cortex.Streams.Kafka/Cortex.Streams.Kafka.csproj b/src/Cortex.Streams.Kafka/Cortex.Streams.Kafka.csproj index bd11617..d3ddb2f 100644 --- a/src/Cortex.Streams.Kafka/Cortex.Streams.Kafka.csproj +++ b/src/Cortex.Streams.Kafka/Cortex.Streams.Kafka.csproj @@ -4,8 +4,8 @@ net8.0 enable - 1.0.1 - 1.0.1 + 2.0.0 + 2.0.0 Buildersoft Cortex Framework Buildersoft Buildersoft,EnesHoxha @@ -16,7 +16,7 @@ https://github.com/buildersoftio/cortex cortex vortex eda streaming distributed streams states kafka pulsar rocksdb - 1.0.1 + 2.0.0 license.md cortex.png Cortex.Streams.Kafka @@ -52,9 +52,9 @@ - - - + + + diff --git a/src/Cortex.Streams.Kafka/KafkaKeyValueSinkOperator.cs b/src/Cortex.Streams.Kafka/KafkaKeyValueSinkOperator.cs new file mode 100644 index 0000000..be127d6 --- /dev/null +++ b/src/Cortex.Streams.Kafka/KafkaKeyValueSinkOperator.cs @@ -0,0 +1,65 @@ +using Confluent.Kafka; +using Cortex.Streams.Kafka.Serializers; +using Cortex.Streams.Operators; +using System; +using System.Collections.Generic; + +namespace Cortex.Streams.Kafka +{ + /// + /// Kafka sink that accepts KeyValuePair so message keys are produced. + /// + public sealed class KafkaSinkOperator : ISinkOperator> + { + private readonly string _bootstrapServers; + private readonly string _topic; + private readonly IProducer _producer; + + public KafkaSinkOperator( + string bootstrapServers, + string topic, + ProducerConfig config = null, + ISerializer keySerializer = null, + ISerializer valueSerializer = null) + { + _bootstrapServers = bootstrapServers ?? throw new ArgumentNullException(nameof(bootstrapServers)); + _topic = topic ?? throw new ArgumentNullException(nameof(topic)); + + var producerConfig = config ?? new ProducerConfig + { + BootstrapServers = _bootstrapServers + }; + + keySerializer ??= new DefaultJsonSerializer(); + valueSerializer ??= new DefaultJsonSerializer(); + + _producer = new ProducerBuilder(producerConfig) + .SetKeySerializer(keySerializer) + .SetValueSerializer(valueSerializer) + .Build(); + } + + public void Process(KeyValuePair input) + { + var msg = new Message { Key = input.Key, Value = input.Value }; + _producer.Produce(_topic, msg, deliveryReport => + { + if (deliveryReport.Error.IsError) + { + Console.WriteLine($"Delivery Error: {deliveryReport.Error.Reason}"); + } + }); + } + + public void Start() + { + // no-op + } + + public void Stop() + { + _producer.Flush(TimeSpan.FromSeconds(10)); + _producer.Dispose(); + } + } +} diff --git a/src/Cortex.Streams.Kafka/KafkaKeyValueSourceOperator.cs b/src/Cortex.Streams.Kafka/KafkaKeyValueSourceOperator.cs new file mode 100644 index 0000000..ee33f96 --- /dev/null +++ b/src/Cortex.Streams.Kafka/KafkaKeyValueSourceOperator.cs @@ -0,0 +1,97 @@ +using Confluent.Kafka; +using Cortex.Streams.Kafka.Deserializers; +using Cortex.Streams.Operators; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Cortex.Streams.Kafka +{ + /// + /// Kafka source that emits KeyValuePair so the pipeline can use message keys. + /// + public sealed class KafkaSourceOperator : ISourceOperator> + { + private readonly string _bootstrapServers; + private readonly string _topic; + private readonly IConsumer _consumer; + private CancellationTokenSource _cts; + private Task _consumeTask; + + + public KafkaSourceOperator(string bootstrapServers, + string topic, + ConsumerConfig config = null, + IDeserializer keyDeserializer = null, + IDeserializer valueDeserializer = null) + { + _bootstrapServers = bootstrapServers ?? throw new ArgumentNullException(nameof(bootstrapServers)); + _topic = topic ?? throw new ArgumentNullException(nameof(topic)); + + var consumerConfig = config ?? new ConsumerConfig + { + BootstrapServers = _bootstrapServers, + GroupId = Guid.NewGuid().ToString(), + AutoOffsetReset = AutoOffsetReset.Earliest, + EnableAutoCommit = true, + }; + + keyDeserializer ??= new DefaultJsonDeserializer(); + valueDeserializer ??= new DefaultJsonDeserializer(); + + _consumer = new ConsumerBuilder(consumerConfig) + .SetKeyDeserializer(keyDeserializer) + .SetValueDeserializer(valueDeserializer) + .Build(); + } + + + public void Start(Action> emit) + { + if (emit == null) throw new ArgumentNullException(nameof(emit)); + + _cts = new CancellationTokenSource(); + _consumer.Subscribe(_topic); + + _consumeTask = Task.Run(() => + { + try + { + while (!_cts.Token.IsCancellationRequested) + { + var result = _consumer.Consume(_cts.Token); + emit(new KeyValuePair(result.Message.Key, result.Message.Value)); + } + } + catch (OperationCanceledException) + { + // shutting down - consume loop canceled + } + finally + { + _consumer.Close(); + } + }, _cts.Token); + } + + public void Stop() + { + if (_cts == null) + return; + + _cts.Cancel(); + try + { + _consumeTask?.Wait(); + } + catch + { + /* swallow aggregate canceled */ + } + + _consumer.Dispose(); + _cts.Dispose(); + } + } +} diff --git a/src/Cortex.Streams.Kafka/KafkaSinkOperator.cs b/src/Cortex.Streams.Kafka/KafkaSinkOperator.cs index abe93af..00608fc 100644 --- a/src/Cortex.Streams.Kafka/KafkaSinkOperator.cs +++ b/src/Cortex.Streams.Kafka/KafkaSinkOperator.cs @@ -5,7 +5,7 @@ namespace Cortex.Streams.Kafka { - public class KafkaSinkOperator : ISinkOperator + public sealed class KafkaSinkOperator : ISinkOperator { private readonly string _bootstrapServers; private readonly string _topic; diff --git a/src/Cortex.Streams.Kafka/KafkaSourceOperator.cs b/src/Cortex.Streams.Kafka/KafkaSourceOperator.cs index 4553e4e..7e668f1 100644 --- a/src/Cortex.Streams.Kafka/KafkaSourceOperator.cs +++ b/src/Cortex.Streams.Kafka/KafkaSourceOperator.cs @@ -7,7 +7,7 @@ namespace Cortex.Streams.Kafka { - public class KafkaSourceOperator : ISourceOperator + public sealed class KafkaSourceOperator : ISourceOperator { private readonly string _bootstrapServers; private readonly string _topic; @@ -15,7 +15,10 @@ public class KafkaSourceOperator : ISourceOperator private CancellationTokenSource _cts; private Task _consumeTask; - public KafkaSourceOperator(string bootstrapServers, string topic, ConsumerConfig config = null, IDeserializer deserializer = null) + public KafkaSourceOperator(string bootstrapServers, + string topic, + ConsumerConfig config = null, + IDeserializer deserializer = null) { _bootstrapServers = bootstrapServers; _topic = topic; @@ -53,7 +56,7 @@ public void Start(Action emit) } catch (OperationCanceledException) { - // Consume loop canceled + // shutting down - consume loop canceled } finally { @@ -64,8 +67,21 @@ public void Start(Action emit) public void Stop() { + if (_cts == null) + return; + _cts.Cancel(); - _consumeTask.Wait(); + try + { + _consumeTask?.Wait(); + } + catch + { + /* swallow aggregate canceled */ + } + + _consumer.Dispose(); + _cts.Dispose(); } } }