tonysneed

docker exec -it broker bash
cd /usr/bin
./kafka-console-consumer --bootstrap-server broker:29092 --topic "processed-events"

tonysneed

git clone https://github.com/confluentinc/cp-all-in-one
cd cp-all-in-one
git checkout 5.5.0-post
cd cp-all-in-one/

docker-compose up -d --build
docker-compose ps

tonysneed

public class Program
{
    public static void Main(string[] args)
    {
        CreateHostBuilder(args).Build().Run();
    }

    public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
            .ConfigureServices((hostContext, services) =>

tonysneed

public class TransformHandler : MessageHandler
{
    public override async Task<Message> HandleMessage(Message sourceMessage)
    {
        var message = (Message<int, string>)sourceMessage;
        var sinkMessage = new Message<int, string>(message.Key, message.Value.ToUpper());
        return await base.HandleMessage(sinkMessage);
    }
}

tonysneed

public static class KafkaUtils
{
    public static IConsumer<int, string> CreateConsumer(string brokerList, List<string> topics)
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = brokerList,
            GroupId = "sample-consumer"
        };
        var consumer = new ConsumerBuilder<int, string>(config).Build();

tonysneed

public class Worker : BackgroundService
{
    private readonly IEventProcessor _eventProcessor;
    private readonly ILogger<Worker> _logger;

    public Worker(IEventProcessor eventProcessor, ILogger<Worker> logger)
    {
        _eventProcessor = eventProcessor;
        _logger = logger;
    }

tonysneed

dotnet add package EventStreamProcessing.Kafka

tonysneed

dotnet new worker --name MyWorker

tonysneed

public override async Task Process(CancellationToken cancellationToken = default)
{
    // Build chain of handlers
    BuildHandlerChain();

    // Consume event
    var sourceEvent = consumer.ConsumeEvent(cancellationToken);

    // Return if EOF
    if (sourceEvent == null) return;

tonysneed

public class Message<TKey, TValue> : Message
{
    public TKey Key { get; set; }
    public TValue Value { get; set; }

    public Message(TKey key, TValue value)
    {
        Key = key;
        Value = value;
    }