In this sample the producer publishes batches of messages that include some tombstones (messages with a key but a null value) and the consumer processes them in batch, telling the tombstones apart from the regular messages.
What This Sample Shows
Publishing a batch with WrapAndPublishBatchAsync, mapping each source item to a message and returning null (every number divisible by 30) to produce a tombstone.
Setting the Kafka key of each envelope (N<number>) so that the tombstones still carry a key.
Consuming in batches of up to 100 messages (or whatever is received within 5 seconds) with EnableBatchProcessing.
Subscribing to a stream of IInboundEnvelope<SampleMessage> and checking IsTombstone to handle the tombstones, using GetKafkaKey to read the key.
How to Run
Start the infrastructure as described in Samples, then run the two applications from the repository root.
dotnet run --project ./samples/Kafka/BatchWithTombstone.Producer/.
dotnet run --project ./samples/Kafka/BatchWithTombstone.Consumer/.
The consumer logs every received message or tombstone together with its key and then the size of each processed batch.
Common
The message being exchanged is defined in a common project.
namespace Silverback.Samples.Kafka.BatchWithTombstone.Common;
public class SampleMessage
{
public int Number { get; set; }
}
using Microsoft.Extensions.DependencyInjection;
using Silverback.Configuration;
using Silverback.Messaging.Configuration;
namespace Silverback.Samples.Kafka.BatchWithTombstone.Producer;
public class Startup
{
public void ConfigureServices(IServiceCollection services)
{
// Enable Silverback
services.AddSilverback()
// Use Apache Kafka as message broker
.WithConnectionToMessageBroker(options => options.AddKafka())
// Delegate the broker clients configuration to a separate class
.AddBrokerClientsConfigurator<BrokerClientsConfigurator>();
// Add the hosted service that produces the random sample messages
services.AddHostedService<ProducerBackgroundService>();
}
public void Configure()
{
}
}
using Silverback.Messaging.Configuration;
using Silverback.Samples.Kafka.BatchWithTombstone.Common;
namespace Silverback.Samples.Kafka.BatchWithTombstone.Producer;
public class BrokerClientsConfigurator : IBrokerClientsConfigurator
{
public void Configure(BrokerClientsConfigurationBuilder builder)
{
builder
.AddKafkaClients(
clients => clients
// The bootstrap server address is needed to connect
.WithBootstrapServers("PLAINTEXT://localhost:19092")
// Add a producer
.AddProducer(
producer => producer
// Produce the SampleMessage to the samples-batch topic
.Produce<SampleMessage>(
endpoint => endpoint
.ProduceTo("samples-batch-with-tombstone"))));
}
}
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Silverback.Messaging.Messages;
using Silverback.Messaging.Publishing;
using Silverback.Samples.Kafka.BatchWithTombstone.Common;
namespace Silverback.Samples.Kafka.BatchWithTombstone.Producer;
public class ProducerBackgroundService : BackgroundService
{
private readonly IPublisher _publisher;
private readonly ILogger<ProducerBackgroundService> _logger;
public ProducerBackgroundService(
IPublisher publisher,
ILogger<ProducerBackgroundService> logger)
{
_publisher = publisher;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
int number = 0;
while (!stoppingToken.IsCancellationRequested)
{
await ProduceMessagesAsync(++number * 100);
await Task.Delay(50, stoppingToken);
}
}
private async Task ProduceMessagesAsync(int number)
{
try
{
IEnumerable<int> batch = Enumerable.Range(number, 100);
await _publisher.WrapAndPublishBatchAsync(
batch,
n => n % 30 == 0 ? null : new SampleMessage { Number = n },
(envelope, n) => envelope.SetKafkaKey($"N{n}"));
_logger.LogInformation("Produced {FirstNumber}-{LastNumber}", number, number + 99);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to produce {FirstNumber}-{LastNumber}", number, number + 99);
}
}
}
using Microsoft.Extensions.DependencyInjection;
using Silverback.Configuration;
using Silverback.Messaging.Configuration;
namespace Silverback.Samples.Kafka.BatchWithTombstone.Consumer;
public class Startup
{
public void ConfigureServices(IServiceCollection services)
{
// Enable Silverback
services.AddSilverback()
// Use Apache Kafka as message broker
.WithConnectionToMessageBroker(options => options.AddKafka())
// Delegate the broker clients configuration to a separate class
.AddBrokerClientsConfigurator<BrokerClientsConfigurator>()
// Register the subscribers
.AddSingletonSubscriber<SampleMessageBatchSubscriber>();
}
public void Configure()
{
}
}
using System;
using Silverback.Messaging.Configuration;
using Silverback.Samples.Kafka.BatchWithTombstone.Common;
namespace Silverback.Samples.Kafka.BatchWithTombstone.Consumer;
public class BrokerClientsConfigurator : IBrokerClientsConfigurator
{
public void Configure(BrokerClientsConfigurationBuilder builder)
{
builder
.AddKafkaClients(
clients => clients
// The bootstrap server address is needed to connect
.WithBootstrapServers("PLAINTEXT://localhost:19092")
// Add a consumer
.AddConsumer(
consumer => consumer
// Set the consumer group id
.WithGroupId("sample-consumer")
// AutoOffsetReset.Earliest means that the consumer
// must start consuming from the beginning of the topic,
// if no offset was stored for this consumer group
.AutoResetOffsetToEarliest()
// Consume the SampleMessage from the samples-batch topic
.Consume<SampleMessage>(
endpoint => endpoint
.ConsumeFrom("samples-batch-with-tombstone")
// Configure processing in batches of 100 messages,
// with a max wait time of 5 seconds
.EnableBatchProcessing(100, TimeSpan.FromSeconds(5)))));
}
}
using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Silverback.Messaging.Messages;
using Silverback.Samples.Kafka.BatchWithTombstone.Common;
namespace Silverback.Samples.Kafka.BatchWithTombstone.Consumer;
public class SampleMessageBatchSubscriber
{
private readonly ILogger<SampleMessageBatchSubscriber> _logger;
public SampleMessageBatchSubscriber(ILogger<SampleMessageBatchSubscriber> logger)
{
_logger = logger;
}
public async Task OnBatchReceivedAsync(IAsyncEnumerable<IInboundEnvelope<SampleMessage>> batch)
{
int count = 0;
await foreach (IInboundEnvelope<SampleMessage> envelope in batch)
{
if (envelope.IsTombstone)
{
_logger.LogInformation(
"Received tombstone message with key {Key}",
envelope.GetKafkaKey());
}
else
{
_logger.LogInformation(
"Received message with key {Key} and number {Number}",
envelope.GetKafkaKey(),
envelope.Message?.Number);
}
count++;
}
_logger.LogInformation("Received batch of {Count} messages", count);
}
}