Transports Kafka

Apache Kafka Transport

Request and response between services over Kafka topics: one request topic for each distributed key, with the partitions and the replication factor you choose.

Installation

dotnet add package FxMap.Kafka

Every service that maps data needs the core FxMap package, and the owning service also needs a data provider. All FxMap.* packages must use the same version.

Configuration

AddKafka takes the address of the brokers, and optionally TLS settings and the number of partitions and the replication factor of the topics FxMap creates. The same call is used on the service that needs data and on the service that owns it:

Basic Configuration

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddProfilesFromAssemblyContaining<SomeProfileAssemblyMarker>();
    cfg.AddKafka(c => c.Host("localhost:9092"));
});

Several Brokers

The host is handed to Kafka as its bootstrap servers, so a list of brokers separated by commas works:

cfg.AddKafka(c => c.Host("kafka1.example.com:9092,kafka2.example.com:9092,kafka3.example.com:9092"));

TLS

Ssl(...) sets the TLS options of the connection. They are the SSL settings of the Kafka client: SslCaLocation, SslCertificateLocation, SslKeyLocation, SslKeyPassword and the others of KafkaSslOptions.

cfg.AddKafka(c =>
{
    c.Host("kafka.example.com:9093");
    c.Ssl(o =>
    {
        o.SslCaLocation = "/certs/ca.pem";
        o.SslCertificateLocation = "/certs/client.pem";
        o.SslKeyLocation = "/certs/client.key";
        o.SslKeyPassword = builder.Configuration["Kafka:KeyPassword"];
    });
});

A process shares one Kafka producer between its client and its server.

Partitions and Replication

Since 4.1.5 you can set the number of partitions and the replication factor of the topics that FxMap creates. Both were fixed at 1 before, which does not fit a cluster of several brokers or several instances of a service.

cfg.AddKafka(c =>
{
    c.Host("localhost:9092");
    c.Partitions(6);          // partitions of the request topic of each distributed key (default 1)
    c.ReplicationFactor(3);   // replication factor of the topics FxMap creates (default 1)
});

• The requests are spread over the partitions, and the instances of a service share them as one consumer group. An instance gets whole partitions, so a topic needs at least as many partitions as the instances that should work at the same time.
• The values apply only when FxMap creates a topic. A topic that exists is not changed: increase its partitions with the tools of Kafka.
• The reply topic of a client is read by one consumer, so it keeps one partition.
• A cluster that has to survive the loss of a broker needs a replication factor of 3, and it cannot be higher than the number of brokers.
• The default stays 1, so nothing changes unless you set the values.

Topics

Requests go to the topic fxmap-request-topic-{distributed key full name} (in lower case), which FxMap creates for you. Each application listens for responses on fxmap-response-topic-[application.friendly.name]. Do not use these topics for anything else.

Server: the service that owns the data

Register the entities and a data provider as usual, and add Kafka. The service consumes the request topic of each distributed key it owns, and no endpoint is needed:

builder.Services.AddFxMap(cfg =>
    {
        cfg.AddEntitiesFromAssemblyContaining<UserConfig>();
        cfg.AddKafka(c => c.Host("localhost:9092"));
    })
    .AddEntityFrameworkCore(c => c.AddDbContexts(typeof(UserDbContext)));

How a request travels

1. Each request holds the ids and the expressions for one distributed key and one level. A hundred objects that need a UserName make one request.
2. The client produces it to the request topic of the key and reads the answer on its response topic.
3. The owner answers with the values, and with the expression errors it isolated (4.1.3+).
4. If nobody answers within the request timeout (30 seconds by default, cfg.SetRequestTimeOut(...)), the request fails. See Retry Policy to retry it.

Complete example

// Shared contract
public sealed class UserDistributedKey : IDistributedKey;
 
// Owner (users service): the entity and how it is exposed
public class UserConfig : EntityConfigureOf<User>
{
    protected override void Configure()
    {
        Id(x => x.Id);
        DefaultProperty(x => x.Name);
        UseDistributedKey<UserDistributedKey>();
    }
}
 
// Client (orders service): the DTO and where its values come from
public class OrderResponse
{
    public string Id { get; set; }
    public string UserId { get; set; }
    public string? UserName { get; set; }
    public string? UserEmail { get; set; }
}
 
public class OrderResponseProfile : ProfileOf<OrderResponse>
{
    protected override void Configure() =>
        UseDistributedKey<UserDistributedKey>()
            .Of(x => x.UserId)
            .For(x => x.UserName)
            .For(x => x.UserEmail, "Email");
}
 
// In an endpoint of the orders service
await distributedMapper.MapDataAsync(orders);

Good to know

Supervised servers

The servers run under the supervisor, which restarts a server that fails. Since 4.1.5 a Kafka server can be started again after it was stopped: it creates its consumer on each start.

Plan the partitions

Use as many partitions as the instances of a service that should work at the same time, and monitor the consumer lag to decide how many instances you need.

Replication in a cluster

Set the replication factor (usually 3) in a cluster, so the topics survive the loss of a broker. Topics that already exist keep their settings.

---

Source: https://fxmapper.net/docs/transports/kafka/ — the website is always the most up to date version.