Akka.Streams.Kafka

By the end of this lesson you will be able to:

• explain what backpressure changes compared with a raw Consume() loop — and what it does not change;
• choose between PlainSource and CommittableSource, and between PlainSink and FlexiFlow, from the delivery guarantee you need;
• wire a consume → transform → produce pipeline whose offsets are committed only after the output write is acknowledged;
• shut a pipeline down with DrainAndShutdown so in-flight work finishes and its offsets are committed;
• trace which offsets are committed, replayed or skipped after a crash, and recognise the five common pitfalls.

Prerequisites: "Raw Confluent.Kafka Client" (the consumer and producer APIs this library wraps), "Consumer & Manual Commits" (what a committed offset means) and "Delivery Semantics & Schema Strategy" (at-least-once versus exactly-once). Basic C# async/await is assumed; no prior Akka experience is needed.

Why Akka.Streams.Kafka: From Raw Consumer to Reactive Pipeline

If you've spent time with Confluent.Kafka's raw consumer API, you know the poll loop intimately. You call consumer.Consume(timeout), hand the result to your processing logic, and repeat — forever. It feels straightforward until the day your downstream database slows down, your enrichment service starts timing out, or a burst of messages arrives faster than your handler can process them. At that point, a question that seemed settled reopens itself: who decides how fast messages flow through your system? With a raw poll loop, the answer is often "nobody in particular" — and that gap is exactly what Akka.Streams.Kafka (the Kafka connector for Akka.Streams, the stream-processing library of the Akka.NET actor toolkit) was designed to close.

The Poll Loop Problem

The standard Confluent.Kafka consumer model looks roughly like this:

using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("orders");

while (!cancellationToken.IsCancellationRequested)
{
    // One message per iteration: the next Consume() only happens
    // after ProcessOrder has returned.
    var result = consumer.Consume(cancellationToken);
    ProcessOrder(result.Message.Value); // Could be slow, could throw
    consumer.Commit(result);
}

This loop is not reckless — it pulls at exactly the rate ProcessOrder completes, and librdkafka's prefetch queue is bounded (queued.max.messages.kbytes, 64 MiB by default). The trouble starts when that rate is not good enough:

• No concurrency. Throughput is capped at one message per handler latency. A 200 ms enrichment call means at most 5 messages per second, however many cores and connections you have.
• One slow call threatens group membership. While ProcessOrder runs, nothing calls Consume(). If a single call takes longer than max.poll.interval.ms (five minutes by default), the client treats itself as failed, leaves the group, and a group rebalance moves its partitions elsewhere.
• Adding concurrency means building it yourself. The moment you hand messages to a Channel<T> and a pool of workers, you own the buffer bound, the overflow policy, the ordering rules, and — hardest of all — the bookkeeping that decides which offset is safe to commit when workers finish out of order.

None of these problems is unsolvable, but each needs coordination machinery that has nothing to do with your business logic. That machinery is what Akka.Streams.Kafka provides.

Backpressure: Downstream Demand Flows Upstream

Backpressure is the mechanism by which a pipeline stage signals to the stage feeding it that it's ready to accept more work. Rather than a producer pushing at maximum rate, a consumer pulls at the rate it can sustain. The key insight is directional: demand signals flow upstream, data flows downstream.

[Kafka Topic] ←── demand signal ──── [Deserializer] ←── demand signal ──── [DB Writer]
     │                                      │                                      │
     └──────────── message data ───────────►└──────────── message data ───────────►

When the DB Writer is busy, it stops requesting new records. The Deserializer, receiving no demand from downstream, stops requesting records from Kafka. The Kafka source then pauses fetching — not because you wrote code to pause it, but because the pipeline's structure enforces it.

It is worth being precise about how. Akka.Streams.Kafka runs the Confluent.Kafka consumer inside an actor (Akka's unit of concurrency: an object that handles the messages in its mailbox one at a time) that polls on a short timer (every 50 ms by default). When no stage is asking for records, that actor calls Pause on the assigned partitions — so no new records are fetched — but it keeps polling. Continuing to poll is what keeps the member inside max.poll.interval.ms and in the group, even while downstream is stuck for minutes.

This isn't a new idea. The Reactive Streams specification formalizes this demand protocol so that libraries from different authors can compose safely. Akka.Streams implements this specification, which means a Kafka source, a database sink, and an HTTP flow written independently can all participate in the same back-pressure contract without custom glue code.

Key Principle: Backpressure doesn't make your pipeline faster — it makes it honest. Instead of hiding the speed mismatch inside a buffer, it surfaces the constraint and lets the slowest stage govern the whole pipeline's throughput.

How Akka.Streams Models a Pipeline

Akka.Streams represents a pipeline as a graph composed of three primary building blocks:

A working pipeline is built by connecting these shapes and then materializing the graph — the step that allocates resources, wires up actors, and starts data flowing. The M in each shape is its materialized value: what running the graph hands back for that stage, such as the Kafka source's IControl or a sink's completion Task. Until materialization, you're working with a blueprint that can be reused (though accidentally materializing it twice is a pitfall covered later in this lesson).

Here's a minimal illustration of shape composition:

// Blueprint construction — nothing runs yet.
var source = KafkaConsumer.PlainSource(
    consumerSettings,
    Subscriptions.Topics("orders")
);

var parseFlow = Flow.Create<ConsumeResult<string, string>>()
    .Select(record => JsonSerializer.Deserialize<OrderEvent>(record.Message.Value));

var logSink = Sink.ForEach<OrderEvent>(evt =>
    Console.WriteLine($"Received order {evt.OrderId}"));

// Materializing the graph — NOW it runs, NOW Kafka polling begins.
var control = source
    .Via(parseFlow)
    .ToMaterialized(logSink, Keep.Left)   // Keep.Left: keep the left (source's) materialized value
    .Run(materializer);   // materializer = system.Materializer(): the engine that runs blueprints
// `control` is IControl — the handle for graceful shutdown, covered in "The IControl Materialized Value" below.

KafkaConsumer.PlainSource
        │
        │  ConsumeResult<string, string>
        ▼
  parseFlow (Select)
        │
        │  OrderEvent
        ▼
  logSink (ForEach)

Every arrow in this diagram is a back-pressured channel. If logSink is busy, parseFlow receives no demand and holds its current element. PlainSource receives no demand from parseFlow, so its consumer actor pauses the partitions (no new fetches) while continuing to poll. The consumer stays a healthy group member because it keeps polling, and it isn't fetching records it can't yet process.

Kafka can participate as any of the three shapes depending on what your pipeline needs:

• As a Source: KafkaConsumer.PlainSource or CommittableSource — reads records and emits them downstream.
• As a Sink: KafkaProducer.PlainSink — accepts ProducerRecord instances (an outgoing record: topic, key and value) as a terminal stage.
• As a Flow: KafkaProducer.FlexiFlow — writes to Kafka and emits a result downstream once the write is acknowledged, carrying a value you passed through (typically the consumer offset), enabling patterns like committing that offset only after a successful produce.

Trade-offs You Accept

Akka.Streams.Kafka is not a free upgrade from raw Confluent.Kafka. The ergonomics come with concrete costs worth naming honestly.

Actor system overhead. Running Akka.Streams requires an ActorSystem — a non-trivial runtime with its own dispatchers (which schedule actors and stages onto threads), configuration (HOCON, Akka's JSON-like config format), and lifecycle. (Akka.NET's default dispatcher runs on the .NET thread pool; Akka.Streams.Kafka's consumer actor and producer stages are configured to use akka.kafka.default-dispatcher, which by default also runs on the .NET thread pool.) For a microservice with simple processing logic and a single topic, this overhead may genuinely not be worth it. Raw Confluent.Kafka with a bounded Channel<T> is often sufficient.

Steeper learning curve. The graph DSL, materialization semantics, supervision strategies (rules for what happens when a stage throws), and dispatcher model are a genuine investment. Developers familiar with LINQ and async/await often find the first week with Akka.Streams disorienting — particularly around how exceptions propagate and how threading is managed.

Reactive streams coordination cost. The demand-signaling protocol adds overhead compared to a tight polling loop. For trivial per-message work at extreme throughput, this overhead can be measurable. Akka.Streams uses fusing — collapsing adjacent stages into a single actor — to reduce this significantly, but it doesn't eliminate it. The practical threshold where ergonomics clearly outweigh the cost is pipelines doing meaningful per-message I/O work, where coordination overhead is negligible relative to I/O latency.

The shift from a raw poll loop to a declarative graph isn't just a style preference — it's a change in where the invariants of your pipeline are enforced. In a poll loop, correct behavior depends on the discipline of the code you write. In a backpressured graph, correct behavior emerges from the structure of the graph itself.

✍️ Exercise: predict the loop's behaviour

A topic receives 50 messages per second. ProcessOrder calls an enrichment API that takes 200 ms per message.

1. How many messages per second does the raw loop above process, and how fast does consumer lag grow?
2. One day the enrichment API hangs for six minutes on a single call. What happens to this consumer's group membership?
3. The same work runs in Akka.Streams.Kafka as SelectAsync(parallelism: 16, ...), and the API can serve 16 concurrent calls at 200 ms each. Can the pipeline keep up?

Core Abstractions: Consumer Sources, Producer Sinks, and the Materialized Control Handle

Akka.Streams.Kafka surfaces exactly three concerns at its API boundary: reading from Kafka, writing to Kafka, and controlling a running stream from outside of it. Understanding what each abstraction emits, accepts, or returns — and when to reach for one variant over another — is the foundation everything else in this library builds on.

Consumer Sources: PlainSource vs. CommittableSource

KafkaConsumer has a dozen factory methods (partitioned sources, external sources, a transactional source and more), but two cover most pipelines, and the choice between them is a delivery-guarantee decision you make at graph construction time, not at runtime.

KafkaConsumer.PlainSource emits ConsumeResult<TKey, TValue> — the same record type you'd receive from a raw Confluent.Kafka poll call. It gives the graph no commit path, and Akka.Streams.Kafka's default consumer configuration sets enable.auto.commit = false, so by default a PlainSource never commits anything. (You can switch librdkafka's auto-commit back on with .WithProperty("enable.auto.commit", "true"), but then offsets are committed on a timer whether or not your stages finished — the at-most-once risk described in "Consumer & Manual Commits".) PlainSource is appropriate when (a) replaying or skipping on restart is acceptable, for example a live dashboard, or (b) you store offsets yourself in an external system and seek to them on startup.

KafkaConsumer.CommittableSource emits CommittableMessage<TKey, TValue>. That wrapper bundles the ConsumeResult together with an ICommittableOffset handle you can pass downstream and commit, individually or in batches. Committing through this handle advances the Kafka consumer group offset in a way your application controls explicitly. This is the entry point for at-least-once delivery: you only commit an offset after your downstream stage has successfully processed the message, so a crash before the commit causes Kafka to re-deliver from the last committed position.

// PlainSource — no commit path; by default nothing is ever committed
var plainSource = KafkaConsumer.PlainSource(
    consumerSettings,
    Subscriptions.Topics("orders")
);
// Emits: ConsumeResult<string, string>

// CommittableSource — explicit offset control for at-least-once delivery
var committableSource = KafkaConsumer.CommittableSource(
    consumerSettings,
    Subscriptions.Topics("orders")
);
// Emits: CommittableMessage<string, string>
// Each element carries msg.Record (the ConsumeResult) and msg.CommitableOffset
// (the library spells it with one "t"). After processing, send the offset to
// Committer.Sink / Committer.Flow — they batch commits for you.

Do not call .Commit() on the offset yourself: the method is marked [Obsolete] with the message "Do not use directly. Use Committer.Flow or Committer.Sink instead", because per-message commits bypass batching and error handling.

Key Principle: For these two sources, the element type tells you the commit contract. ConsumeResult from PlainSource = no commit path exists in the graph. CommittableMessage = a commit path exists and you are responsible for exercising it. (SourceWithOffsetContext is the exception worth knowing: it emits ConsumeResult but carries the committable offset alongside as context.)

⚠️ Switching from CommittableSource to PlainSource to simplify a pipeline silently drops at-least-once guarantees. With the default configuration nothing is committed any more, so after a restart the consumer resumes from the group's last committed offset from before the switch and replays everything since — or, if the group has never committed, from auto.offset.reset, whose librdkafka default of latest skips the backlog entirely. Either way it surfaces as silent duplication or silent loss, not an exception.

Subscription Strategies and Offset-Reset Behavior

Both source factories accept an ISubscription object that controls how the consumer is assigned to partitions. The Subscriptions static class provides four factory methods, each implying different behavior around partition assignment and offset initialization.

Topic-name subscription (Subscriptions.Topics("orders", "payments")) joins the consumer group, whose coordinator assigns partitions dynamically and rebalances when group membership changes. When the group has no committed offset for a partition, auto.offset.reset decides where to start; set it with .WithProperty("auto.offset.reset", "earliest") (or through WithConsumerConfig), because librdkafka's default is latest.

Pattern subscription (Subscriptions.TopicPattern("^order\\..*")) works like topic-name subscription, but the consumer subscribes to every topic whose name matches a regular expression. Two details bite: Confluent.Kafka only treats the string as a regex when it starts with ^ (without it, "order.*" is taken as a literal topic name), and where the matching happens depends on the group protocol: under the classic protocol (librdkafka's current default) the client matches against cluster metadata, while under group.protocol=consumer (KIP-848) the broker matches, and the pattern must match the whole topic name (^order.*, not ^order). New topics matching the pattern are picked up without a code change.

Explicit partition assignment (Subscriptions.Assignment(new TopicPartition("orders", 0))) pins your application to specific partitions, with no group-managed rebalancing: if the instance dies, nobody takes over its partitions. Offsets still work normally — with a group.id, the consumer starts from the group's committed offset for that partition (or auto.offset.reset if there is none) and can commit. When you want to choose the start position yourself, use Subscriptions.AssignmentWithOffset.

// Explicit partition + starting offset: start reading partition 0 from offset 500
var assignmentSub = Subscriptions.AssignmentWithOffset(
    new TopicPartitionOffset("orders", partition: 0, offset: new Offset(500))
);

var source = KafkaConsumer.PlainSource(consumerSettings, assignmentSub);

Mental Model: Think of the four strategies as a dial from fully managed (topic names or a pattern) to fully manual (explicit assignment with offset). As you turn the dial toward manual, you gain precision and give up automatic rebalancing.

Producer Sinks and FlexiFlow

On the production side, the library offers two complementary shapes depending on whether Kafka writing is the terminal step of your graph or a midpoint you need to pass through.

KafkaProducer.PlainSink is a terminal Sink<ProducerRecord<TKey, TValue>, Task<Done>> (Done is Akka's "finished, no value" marker). Use this when producing to Kafka is the final action of the pipeline and you don't need the delivery receipt downstream.

KafkaProducer.FlexiFlow<TKey, TValue, TPassThrough> is a pass-through Flow<IEnvelope<TKey, TValue, TPassThrough>, IResults<TKey, TValue, TPassThrough>, NotUsed> (NotUsed: the stage has no materialized value). It accepts envelopes — usually built with ProducerMessage.Single(record, passThrough) — writes the record, and once Kafka acknowledges the write emits a result that carries your pass-through value back out (result.PassThrough). For a single record the result is a Result whose Metadata is Confluent.Kafka's delivery report, with the partition and offset Kafka assigned. Use FlexiFlow when you need to act after the write is acknowledged — typically committing the consumer offset you passed through, or auditing where the record landed.

            PlainSink                          FlexiFlow

records ──► [KafkaProducer] ──► (done)   envelopes ──► [KafkaProducer] ──► results
             terminal sink                 pass-through; each result carries
                                           your PassThrough value + delivery report

The practical difference is clearest in a consume-transform-produce pipeline with explicit offset commits. PlainSink loses the signal for when the Kafka write succeeded, so you cannot safely commit the consumer offset only after producer confirmation. FlexiFlow gives you that acknowledgment in the stream.

// Consume → produce → commit, with the commit only after Kafka acknowledged the write.
var (control, streamCompletion) = KafkaConsumer
    .CommittableSource(consumerSettings, Subscriptions.Topics("orders"))
    .Select(msg => ProducerMessage.Single(
        new ProducerRecord<string, string>("results", msg.Record.Message.Key, msg.Record.Message.Value),
        passThrough: msg.CommitableOffset))            // carry the offset through the write
    .Via(KafkaProducer.FlexiFlow<string, string, ICommittableOffset>(producerSettings))
    .Select(ICommittable (result) => result.PassThrough) // the offset, now safe to commit
    .ToMaterialized(Committer.Sink(committerSettings), Keep.Both)
    .Run(materializer);

The third type argument of FlexiFlow is the pass-through type — here the ICommittableOffset. The explicit ICommittable return type on the last lambda is needed because Committer.Sink accepts ICommittable elements (the common interface of a single offset and a batch of offsets). committerSettings is created with CommitterSettings.Create(system) and controls how commits are batched. Keep.Both keeps both materialized values — the source's IControl and the sink's Task<Done> — as the pair deconstructed into (control, streamCompletion).

The IControl Materialized Value

Every consumer source in Akka.Streams.Kafka materializes an IControl object — the bridge between your declarative, running graph and the imperative world outside it.

IControl exposes four members:

• Stop() — the source stops emitting new messages. It does not close the Kafka consumer, so offsets for elements already in the stream can still be committed.
• Shutdown() — closes the consumer, after waiting for outstanding commit requests.
• DrainAndShutdown(streamCompletion) — the safe choice for graceful shutdown: calls Stop(), waits for the stream-completion task you pass in (so every in-flight element reaches the end of the graph — for a Committer.Sink, that includes committing its offsets), then calls Shutdown().
• IsShutdown — a Task that completes once the consumer has been closed, whether by Shutdown(), downstream cancellation or a failure.

Because DrainAndShutdown needs the stream-completion task, the library offers DrainingControl<T>, which bundles the two materialized values. The pattern from the library's own documentation:

DrainingControl<Done> control = KafkaConsumer
    .CommittableSource(consumerSettings, Subscriptions.Topics("orders"))
    .Via(processingFlow)   // emits each message's offset as an ICommittable once processed
    .ToMaterialized(Committer.Sink(committerSettings), Keep.Both)
    .MapMaterializedValue(tuple => DrainingControl.Create(tuple.Item1, tuple.Item2))
    .Run(materializer);

Wire it to shutdown somewhere that is actually awaited. In a .NET generic host that is IHostedService.StopAsync:

public async Task StopAsync(CancellationToken cancellationToken)
{
    if (_control is not null)
        await _control.DrainAndShutdown(); // stop, let in-flight work finish and commit, close the consumer
}

(An async lambda registered on ApplicationStopping or ProcessExit is async void: the host does not wait for it, so the process can exit half-way through the drain.)

You can also use IsShutdown as a liveness signal: if it has already completed when your health check runs and no shutdown was requested, the consumer is gone and a restart is warranted.

Actor System Lifecycle and In-Flight Messages

Because Akka.Streams runs on top of an Akka actor system, the stream's runtime is coupled to that system's lifecycle. If you call system.Terminate() while messages are still in flight, the stream stages receive no graceful shutdown signal — the underlying actor machinery simply stops.

Messages that have been fetched from Kafka but whose offsets were not yet committed are re-delivered from the last committed offset on restart. That is correct at-least-once behaviour, but only if you're using CommittableSource faithfully.

system.Terminate() without draining first
        │
        ▼
stream stages stop; fetched-but-uncommitted elements are abandoned
        │
        ▼
on restart, Kafka re-delivers from the last committed offset

The correct shutdown sequence:

await control.DrainAndShutdown(); // 1. stop the source, let in-flight elements finish and commit, close the consumer
await system.Terminate();         // 2. only now tear down the actor system

⚠️ Calling system.Terminate() in a finally block without first awaiting DrainAndShutdown means every element that was in flight — and every offset the Committer.Sink had batched but not yet sent — is processed again on the next startup. With a large parallelism and a long commit interval, that window can be surprisingly large.

Abstractions at a Glance

These five building blocks are the primary vocabulary of any Akka.Streams.Kafka pipeline. Everything else — commit batching, partitioned sources, restart supervision — builds on top of these primitives.

✍️ Exercise: pick the shapes

For each pipeline, choose the source, the producer stage (if any) and how offsets are handled.

1. Read payments, call a fraud-scoring API, write the verdict to payment-verdicts. A crash must never skip a payment.
2. A load-test dashboard counts messages on clicks. After a restart it should simply show current traffic; replays and gaps don't matter.
3. Each event is projected into PostgreSQL, and the partition offset is stored in the same database transaction as the projection.

Wiring a Basic End-to-End Pipeline in C#

With the conceptual scaffolding in place, it's time to assemble a real pipeline and watch the pieces click together. This section builds a complete consume-transform-produce pipeline from scratch: consuming from one Kafka topic, applying an async transformation, and producing to another.

Configuring ConsumerSettings and ProducerSettings

Before any graph can run, Akka.Streams.Kafka needs to know how to reach the broker, which consumer group to join, and how to serialize and deserialize messages. These concerns live in ConsumerSettings<TKey, TValue> and ProducerSettings<TKey, TValue> — immutable configuration objects you construct once and pass into graph stages.

using Akka.Actor;
using Akka.Streams.Kafka.Dsl;
using Akka.Streams.Kafka.Helpers;
using Akka.Streams.Kafka.Messages;
using Akka.Streams.Kafka.Settings;
using Akka.Streams;
using Akka.Streams.Dsl;
using Confluent.Kafka;

// Bootstrap the actor system — streams live inside this system.
var system = ActorSystem.Create("order-pipeline");
var materializer = system.Materializer();

// The key/value types are fixed at compile time; the (de)serializer instances are passed here.
var consumerSettings = ConsumerSettings<Null, string>
    .Create(system, null, Deserializers.Utf8)
    .WithGroupId("order-processor")
    .WithBootstrapServers("localhost:9092")
    .WithProperty("auto.offset.reset", "earliest");

// Each materialized producer stage creates ONE IProducer from these settings and
// reuses it for every message. To share a single producer across several stages,
// build it once with producerSettings.CreateKafkaProducer() and pass it to the
// PlainSink(settings, producer) / FlexiFlow(settings, producer) overloads.
var producerSettings = ProducerSettings<Null, string>
    .Create(system, null, Serializers.Utf8)
    .WithBootstrapServers("localhost:9092");

A few points worth noting. The key and value types are part of the generic type parameters, so a mismatch is a compile error; the (de)serializer objects themselves are ordinary runtime arguments. The .WithProperty() method passes a property straight through to the Confluent.Kafka client (in HOCON these live in the kafka-clients section), so any Confluent.Kafka/librdkafka property name is valid. Akka.Streams.Kafka's reference configuration sets enable.auto.commit = false there by default. Prefer loading as much as possible from HOCON configuration and using .With*() fluent overrides only for values that genuinely vary at runtime.

Building the Graph: Source → Flow → Sink

┌───────────────────────────────────────────────────────────────────┐
│                        Akka.Streams Graph                         │
│                                                                   │
│  KafkaConsumer          Flow                  KafkaProducer       │
│  ┌─────────────┐    ┌──────────────┐         ┌──────────────┐     │
│  │ PlainSource │───▶│ SelectAsync  │────────▶│  PlainSink   │     │
│  │ (orders-in) │    │ (enrich msg) │         │ (orders-out) │     │
│  └─────────────┘    └──────────────┘         └──────────────┘     │
│        │                                             │            │
│        │◀──── backpressure demand signal ────────────│            │
└───────────────────────────────────────────────────────────────────┘

Here is the complete pipeline wired in code:

var (control, streamCompletion) = KafkaConsumer
    .PlainSource(consumerSettings, Subscriptions.Topics("orders-in"))
    // SelectAsync runs up to `parallelism` transforms concurrently.
    // Keeping this async is critical — see the pitfalls section.
    .SelectAsync(parallelism: 4, async message =>
    {
        // Simulate an I/O-bound enrichment step, e.g., a database lookup.
        var enriched = await EnrichOrderAsync(message.Message.Value);

        return new ProducerRecord<Null, string>(
            topic: "orders-out",
            value: enriched
        );
    })
    .ToMaterialized(
        KafkaProducer.PlainSink(producerSettings),
        Keep.Both   // retain both IControl and the sink's completion Task
    )
    .Run(materializer);

static async Task<string> EnrichOrderAsync(string rawOrder)
{
    await Task.Delay(10); // replace with actual DB/HTTP call
    return $"ENRICHED:{rawOrder}";
}

The Keep.Both argument to ToMaterialized returns two useful values: the IControl handle from the consumer source and the sink's Task<Done>, which completes when the stream finishes or faults.

This example uses PlainSource, which does not commit offsets. As established in the previous section, PlainSource is appropriate when offset management is external or when you're experimenting; for production at-least-once delivery, swap in CommittableSource and replace PlainSink with FlexiFlow followed by Committer.Sink, as in the consume → produce → commit example above.

Using SelectAsync for I/O-Bound Work

SelectAsync is the correct choice any time the transformation involves I/O — HTTP calls, database reads, cache lookups. The parallelism parameter sets how many async Tasks can be in-flight simultaneously from this stage.

Key Principle: The parallelism factor is not a thread count — it is a concurrency ceiling on async Tasks. Because these are genuinely async operations, four concurrent enrichment calls typically consume zero threads while awaiting I/O. Tuning it means finding the point at which adding more concurrency stops increasing throughput (or starts overloading the downstream service).

⚠️ Calling .Result or .Wait() inside a synchronous Select instead of using SelectAsync blocks a thread for the duration of each I/O operation and serialises the stage to one call at a time. This pitfall is covered in depth in the next section.

Wiring IControl to a Shutdown Hook

In a console app, turn Ctrl+C into a signal you can await, and do the shutdown work on the main path — not inside an event handler that nothing waits for:

var shutdownRequested = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
Console.CancelKeyPress += (_, e) =>
{
    e.Cancel = true;                  // don't let the runtime kill the process; we'll exit ourselves
    shutdownRequested.TrySetResult();
};

// Run until Ctrl+C, or until the stream ends on its own (for example, a failure).
await Task.WhenAny(shutdownRequested.Task, streamCompletion);

try
{
    // Stop the source, wait for in-flight elements to reach the sink, then close the consumer.
    // If the stream failed, this rethrows its exception (after closing the consumer).
    await control.DrainAndShutdown(streamCompletion);
}
finally
{
    await system.Terminate(); // runs even when the drain rethrows
}

The ordering — drain first, then terminate the actor system — ensures every message the source had already emitted has had the opportunity to pass through every stage before the runtime disappears. This example uses PlainSource, so nothing is committed; with CommittableSource and Committer.Sink, draining is also what gets the final batch of offsets committed.

In a Kubernetes deployment, a pod receives SIGTERM (then SIGKILL after the grace period, 30 seconds by default). Console.CancelKeyPress does not see SIGTERM, so services should use the generic host: it translates SIGTERM into a call to every IHostedService.StopAsync, which is where DrainAndShutdown belongs (shown in the previous section). A member that shuts down this way finishes its in-flight work, commits it (if it commits at all), and closes its consumer so the group can reassign its partitions straight away instead of waiting for a session timeout.

Running Locally and Observing the Pipeline

The fastest way to verify the pipeline behaves as expected is against a single-node Kafka broker. Docker Compose works well here — partition assignment, consumer group coordination, and offset tracking all function identically on a single node.

Once the application is running, verify three things:

1. Partition assignment. The broker's coordinator logs will show the consumer group forming and partitions being assigned.

2. Message flow. Use kafka-console-producer to inject messages into orders-in, then watch orders-out:

# Produce test messages to orders-in
kafka-console-producer --bootstrap-server localhost:9092 --topic orders-in
>order-001
>order-002

# In another terminal, consume from orders-out to verify transformation
kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic orders-out --from-beginning
# Expected output:
# ENRICHED:order-001
# ENRICHED:order-002

3. Consumer group lag. With PlainSource, offsets are never committed, so the tool has no committed position to measure lag from — a correct and expected observation that reinforces why PlainSource fits only replay-tolerant or externally managed offset scenarios.

kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group order-processor \
  --describe

# Sample output:
# GROUP           TOPIC      PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processor orders-in  0          -               47              -

The - in CURRENT-OFFSET (and therefore in LAG) with PlainSource confirms no offsets are being committed. If you commit instead (CommittableSource, then FlexiFlow into Committer.Sink), real numbers will appear in those columns each time the committer flushes a batch.

Consumer lag is one of the most actionable Kafka health signals available. A lag value that grows monotonically means your pipeline's processing throughput is below the producer's publish rate — which in Akka.Streams terms usually means either the SelectAsync parallelism ceiling is too low, or the downstream sink is the bottleneck. Backpressure does not make lag go away; it makes sure the backlog waits in Kafka rather than in your process's memory.

✍️ Exercise: fix this shutdown code

A teammate's service consumes with CommittableSource and commits through Committer.Sink. They report that every deploy reprocesses a few hundred orders. Find three problems.

var (control, done) = KafkaConsumer
    .CommittableSource(consumerSettings, Subscriptions.Topics("orders"))
    .SelectAsync(4, async msg =>
    {
        await HandleAsync(msg.Record.Message.Value);
        return (ICommittable)msg.CommitableOffset;
    })
    .ToMaterialized(Committer.Sink(committerSettings), Keep.Both)
    .Run(materializer);

AppDomain.CurrentDomain.ProcessExit += async (_, _) =>
{
    await control.Stop();
    await system.Terminate();
};

Common Pitfalls When Starting with Akka.Streams.Kafka

Adopting Akka.Streams.Kafka after working with raw Confluent.Kafka is not a drop-in migration — it is a shift in mental model. The mistakes covered here share a common trait: they do not cause immediate, loud failures. Instead, they produce subtle misbehavior — stalled pipelines, silently re-processed messages, duplicate consumers, or streams that die on a single bad payload.

Pitfall 1: Blocking Inside a Stream Stage

The most frequent mistake made by developers coming from imperative Confluent.Kafka code is treating a stream stage as a thread. In a raw consumer loop, calling an async method synchronously with .Result or .Wait() is ugly but functional — the loop owns its thread. Inside an Akka.Streams stage, that assumption breaks.

Stream stages execute on dispatcher threads — by default, the shared .NET thread pool — and adjacent synchronous stages are fused into one actor that processes one element at a time. When a synchronous Select calls .Result on a Task, it holds that thread for the full duration of the call, and the fused section can do nothing else meanwhile. The result is not an error but a collapse in throughput: exactly one HTTP call in flight at a time, so throughput is capped at one message per round trip. Run several such streams, or block in many actors at once, and you also start starving the thread pool that the rest of the application depends on.

// ❌ WRONG: blocking inside a synchronous Select stage
var badStream = KafkaConsumer
    .PlainSource(consumerSettings, Subscriptions.Topics("orders"))
    .Select(msg =>
    {
        // Blocks a thread for the full HTTP round trip; one call in flight at a time.
        var result = httpClient.PostAsync("/process", content).Result;
        return result.StatusCode;
    })
    .RunWith(Sink.Ignore<HttpStatusCode>(), materializer);

// ✅ CORRECT: non-blocking async work via SelectAsync
var goodStream = KafkaConsumer
    .PlainSource(consumerSettings, Subscriptions.Topics("orders"))
    .SelectAsync(parallelism: 4, async msg =>
    {
        // The thread is released while the HTTP call is in flight; up to 4 run concurrently.
        var response = await httpClient.PostAsync("/process", content);
        return response.StatusCode;
    })
    .RunWith(Sink.Ignore<HttpStatusCode>(), materializer);

⚠️ The same problem applies to custom GraphStage implementations (hand-written stream stages) that call blocking APIs in their OnPush handlers (called for each incoming element). If you are writing a custom stage, ensure any async work uses GetAsyncCallback to safely re-enter the stage after a Task completes.

Pitfall 2: Forgetting That PlainSource Does Not Commit Offsets

PlainSource is attractive for its simplicity. Developers sometimes switch to it to reduce boilerplate, then carry that choice forward into production code.

The distinction is not about what Kafka tracks but about what gets written back. PlainSource with enable.auto.commit = false (the default in Akka.Streams.Kafka's ConsumerSettings) means offsets are never committed. With enable.auto.commit = true, commits happen on the client's timer regardless of whether your processing succeeded — adding silent loss to silent duplication.

Consumer restart with PlainSource (auto-commit off):

  Partition offset log:
  [0] [1] [2] [3] [4] [5] [6] [7]
               ^               ^ last produced
               |
        Committed offset = 3 (next to read)
        written before the switch to PlainSource;
        PlainSource never advances it

  On restart: consumer reads from offset 3 again.
  Messages 3-7 are reprocessed silently.

Switching from CommittableSource to PlainSource for simplicity is safe only in read-once, throwaway pipelines. Any pipeline that must survive restarts needs its progress saved: CommittableSource with commits through Committer, or PlainSource with offsets stored in your own database and read back on startup (as in the pick-the-shapes exercise).

Pitfall 3: Accidentally Materializing the Same Blueprint Multiple Times

In Akka.Streams, a graph blueprint is an immutable description of a pipeline — not the pipeline itself. Calling .Run() on a blueprint materializes it, creating a live, running pipeline with its own actor hierarchy, Kafka consumer instance, and consumer group membership.

If you hold a reference to the blueprint and call .Run() more than once — inside a retry loop, a controller action, or a startup method invoked twice — you create multiple independent consumers in the same consumer group:

Single blueprint materialized twice — what Kafka sees:

  Consumer Group: "order-processor"
  ┌─────────────────────────────────────────────────┐
  │  Instance A (materialization 1)                 │
  │    Assigned partitions: [0, 1]                  │
  │                                                 │
  │  Instance B (materialization 2) ← unintended    │
  │    Assigned partitions: [2, 3]                  │
  └─────────────────────────────────────────────────┘
  
  Effect: an unplanned second consumer takes half the partitions; if
  shutdown drains only one IControl, the other's in-flight work is abandoned.
  Starting or stopping either one triggers another rebalance.

// ProcessAsync: Task<ConsumeResult<string, string>> ProcessAsync(ConsumeResult<string, string> msg)

// ❌ WRONG: blueprint stored and Run() called twice
var blueprint = KafkaConsumer
    .PlainSource(consumerSettings, Subscriptions.Topics("orders"))
    .SelectAsync(4, ProcessAsync)
    .ToMaterialized(Sink.Ignore<ConsumeResult<string, string>>(), Keep.Left);

var control1 = blueprint.Run(materializer); // consumer 1 joins group
var control2 = blueprint.Run(materializer); // consumer 2 joins group — unintended!

// ✅ CORRECT: materialize exactly once, hold the IControl reference
var control = KafkaConsumer
    .PlainSource(consumerSettings, Subscriptions.Topics("orders"))
    .SelectAsync(4, ProcessAsync)
    .ToMaterialized(Sink.Ignore<ConsumeResult<string, string>>(), Keep.Left)
    .Run(materializer);

Pro Tip: Guard materialization with a flag or structured startup sequence. In a hosted service, materialize in StartAsync and call DrainAndShutdown in StopAsync.

Pitfall 4: One Slow Element Stalls an Ordered SelectAsync

A pipeline that runs smoothly in integration tests can appear to stall in production — emitting nothing for tens of seconds, then bursting. The usual cause is not a buffer setting but head-of-line blocking: SelectAsync runs up to parallelism tasks concurrently, but it emits their results in input order. If one task is slow, every result behind it waits, even the ones that have already finished.

SelectAsync(parallelism: 4) — results must leave in input order

  offset 100  ██████████████████████████████  (HTTP call hits its 30 s timeout)
  offset 101  ██  done — waiting for 100
  offset 102  ██  done — waiting for 100
  offset 103  ███ done — waiting for 100

  → all 4 slots are occupied, nothing is emitted, no demand reaches the source
  → the consumer pauses its partitions; lag grows at the producers' rate
  → at 30 s the lambda catches 100's timeout and handles it (see below);
    100–103 leave at once and the backlog drains (burst)

What to do about it:

• Bound every call. A per-call timeout (for example HttpClient.Timeout or a CancellationTokenSource) caps how long one element can hold the stage. Pick it from the latency you can tolerate, not the library default. Catch the resulting exception inside the lambda — left uncaught, it fails the whole stream, because SelectAsync's default supervision decision is Stop (see Pitfall 5) — and write the record to a dead-letter topic (a separate topic for records that could not be processed, to retry or inspect later) before the lambda returns. With CommittableSource, returning a plain fallback value instead lets Committer.Sink commit the message's offset although its work never happened, so it is silently skipped.
• Raise parallelism so that more of the following elements finish while the slow one runs. It does not shorten the stall — ordered emission still waits for the slow element — but the backlog clears faster afterwards.
• Use SelectAsyncUnordered only when nothing downstream depends on order. It emits each result as soon as it finishes. With CommittableSource it is dangerous: Committer.Sink commits whatever offset it saw last for a partition, so a later offset that finishes first can be committed while an earlier one is still running — and a crash then skips the unfinished message.

Buffer settings are rarely the culprit. For reference: librdkafka's fetch.min.bytes defaults to 1 (the broker answers as soon as any data exists) and max.partition.fetch.bytes to 1 MiB, which librdkafka raises automatically if a single message is larger. Akka.Streams' max-input-buffer-size (default 16 elements) only applies at asynchronous boundaries inside the graph (points, marked with .Async(), where the graph is split into separately running parts with a buffer between them); it must be a power of two, and a value such as 20 is rejected with an ArgumentException when the materializer settings are created — it is not rounded.

Pitfall 5: Not Handling Failures with RestartSource or Supervision

Akka.Streams propagates failures as stream termination signals — when an unhandled exception escapes a stage, it tears down the entire pipeline. A single malformed message can kill the consumer, stop all partition processing, and potentially leave offsets un-committed.

Two complementary tools prevent this: supervision strategies and RestartSource. They solve different problems.

A supervision strategy intercepts exceptions thrown within a stage and decides whether to Stop the stream, Resume (drop the offending element and continue), or Restart the stage. Supervision is the right tool for per-message errors like deserialization failures.

// Akka.Actor also defines Decider and Directive, so alias the stream versions.
using Decider = Akka.Streams.Supervision.Decider;
using Directive = Akka.Streams.Supervision.Directive;

Decider decider = ex => ex is JsonException
    ? Directive.Resume   // drop the bad message and continue
    : Directive.Stop;    // unknown exceptions still bring down the stream

var done = KafkaConsumer
    .PlainSource(consumerSettings, Subscriptions.Topics("orders"))
    .SelectAsync(4, msg => DeserializeAndProcessAsync(msg))
    .WithAttributes(ActorAttributes.CreateSupervisionStrategy(decider)) // applies to the SelectAsync stage
    .RunWith(Sink.Ignore<ConsumeResult<string, string>>(), materializer);

The supervision strategy is an attribute of the graph (or of individual stages), not of the materializer; to make it the default for everything a materializer runs, create the materializer with ActorMaterializerSettings.Create(system).WithSupervisionStrategy(decider).

⚠️ Directive.Resume silently drops the element — and with CommittableSource, the dropped message's offset never reaches the committer while later offsets do, so it is committed past and never seen again. If the bad message must be preserved, catch the exception inside the SelectAsync body and write the record to a dead-letter topic there. The decider cannot do it for you: it receives only the exception, not the element.

RestartSource is the outer defense: it wraps a source factory and recreates the source with exponential back-off. RestartSource.WithBackoff restarts when the inner source fails or completes; RestartSource.OnFailuresWithBackoff restarts only on failure. Use it for failures of the source itself, not per-message logic errors. Note that a broker outage does not fail the source: its built-in supervision (DefaultConsumerDecider) resumes on non-fatal broker and network errors, and librdkafka keeps reconnecting for as long as the outage lasts. The opt-in connection-checker setting, meant to fail the source when brokers stay unreachable, does not do so in .NET: it only treats a TimeoutException as a failed check, and Confluent.Kafka reports metadata timeouts as KafkaException. If a long outage matters, detect it yourself, for example by alerting on consumer lag. The source fails, and RestartSource recreates it, when that decider says Stop: on fatal client errors, on exceptions from the configured key or value deserializer, and on exceptions that are not Kafka errors. A restart does not get past a record that cannot be deserialized — the new consumer resumes from the last committed offset, reaches the same record and fails again — so consume strings or bytes and deserialize inside the graph, as in the supervision example above.

There is a catch: a restarting source materializes NotUsed, so the IControl of the Kafka source inside it is not returned to you — and without it there is no DrainAndShutdown. One way to keep graceful shutdown is to capture the control of the current incarnation inside the factory and use OnFailuresWithBackoff, so that stopping the source (which completes it) does not trigger a restart:

IControl? currentControl = null; // replaced on every restart

var restartingSource = RestartSource.OnFailuresWithBackoff(
    () => KafkaConsumer
        .PlainSource(consumerSettings, Subscriptions.Topics("orders"))
        .MapMaterializedValue(c => { Volatile.Write(ref currentControl, c); return NotUsed.Instance; }),
    RestartSettings.Create(
        minBackoff: TimeSpan.FromSeconds(3),
        maxBackoff: TimeSpan.FromSeconds(30),
        randomFactor: 0.2)); // restarts are unlimited unless you call WithMaxRestarts

Task<Done> done = restartingSource
    .SelectAsync(4, ProcessAsync)
    .RunWith(Sink.Ignore<ConsumeResult<string, string>>(), materializer);

// On shutdown:
var control = Volatile.Read(ref currentControl);
if (control is not null)
    await control.DrainAndShutdown(done);

(The older overload that takes minBackoff, maxBackoff, randomFactor and maxRestarts directly still compiles but is marked [Obsolete] in favour of RestartSettings.)

Note that RestartSource recreates the Kafka consumer on each restart; the new consumer rejoins the group and resumes from the last committed offset — another reason the PlainSource vs. CommittableSource decision matters so much for resilience.

Key Principle: Layer your failure handling: use supervision strategies for per-message exceptions, and RestartSource for source-level failures. Using only one leaves either message-level or infrastructure-level failures unhandled.

✍️ Exercise: what gets committed after the crash?

A pipeline reads one partition with CommittableSource → SelectAsync(parallelism: 4) → Committer.Sink, using the committer defaults (max-batch 1000, max-interval 10 s: it sends a commit once 1,000 offsets have collected or when its 10 s timer fires, whichever comes first). At t = 0 s the committer has just flushed: the committed offset is 100 (the next offset to read), and its timer next fires at t = 10 s. Between t = 0 and t = 4 s the pipeline finishes offsets 100–139 in order. Then offset 140's HTTP call hangs, while 141, 142 and 143 finish inside SelectAsync. At t = 6 s the pod is killed with SIGKILL.

1. From which offset does the replacement consumer resume, and which messages are handled twice?
2. What changes if the committer is configured with WithMaxInterval(TimeSpan.FromSeconds(1))?
3. Harder: with the 1-second interval, a teammate swaps SelectAsync for SelectAsyncUnordered "to get rid of the stall". What can now go wrong?

Summary and What Comes Next

What Changed in Your Mental Model

The shift from raw Confluent.Kafka to Akka.Streams.Kafka isn't primarily about syntax. It's about where control over consumption rate lives.

In an imperative poll loop, the loop itself is the rate controller: one Consume() per iteration, so throughput is capped by per-message latency, and any concurrency — worker pools, buffers, out-of-order offset tracking — is code you write and must get right. A single long call also puts group membership at risk through max.poll.interval.ms.

Akka.Streams.Kafka replaces this with a graph that enforces backpressure structurally. Downstream stages signal demand upstream through the reactive streams protocol, and the Kafka consumer actor fetches only for partitions that have demand, pausing the rest while it keeps polling to stay in the group. Consumption rate becomes a function of actual downstream capacity, and concurrency is a single parallelism number.

Imperative model:
  Consume() → process → Consume() → ...
  [Rate: one message per handler latency; concurrency is hand-built]

Reactive graph model:
  Sink demands element
    → Flow passes the demand upstream
      → Source fetches from Kafka only then (paused partitions otherwise)
        → records flow back down: Source → Flow → Sink
  [Rate: downstream capacity; concurrency = SelectAsync parallelism]

Three Structural Decisions Made at Graph Construction Time

1. Source type encodes your delivery guarantee. The choice between PlainSource and CommittableSource is not revisable at runtime — it changes the element type flowing through the entire graph. Before writing any graph, answer two questions: must every message be processed at least once, even across a crash? and where is progress (the committed offset) stored?

• At least once, with progress in Kafka: use CommittableSource and commit through Committer.
• At least once, with progress in your own database, written in the same transaction as the results: a PlainSource-family source that reads its start offsets from that database.
• Replays and gaps both acceptable (a live dashboard): PlainSource with no commits.
• If exactly-once is required for a Kafka-to-Kafka pipeline: Akka.Streams.Kafka does not support it yet. KafkaConsumer.TransactionalSource and KafkaProducer.TransactionalFlow/TransactionalSink exist, but the producer side is marked [InternalApi] (not a supported public API) and the library's README says transactions are not implemented yet (issue #85). Use at-least-once with idempotent handling, or Confluent.Kafka's own transactional producer (the Kafka transactions described in "Delivery Semantics & Schema Strategy"). Side effects outside Kafka — database writes, HTTP calls — are never covered by a Kafka transaction and still need idempotent handling.

2. IControl is the imperative escape hatch — keep it narrow. It exposes Stop() (stop emitting), Shutdown() (close the consumer), DrainAndShutdown(streamCompletion) (stop, wait for in-flight work, then close — the graceful shutdown) and IsShutdown (a Task usable as a liveness signal). It is for interacting with a running stream from outside the graph. It is not for driving stream logic from within the graph.

3. Blueprint ≠ instance. A graph definition is a blueprint. Calling .Run() materializes it, creating actual resources. Calling .Run() twice on the same blueprint creates two independent consumers. This is correct when you want parallelism; it is a costly mistake when accidental.

Pitfall Summary

The Canonical Shutdown Pattern

This structure should appear in every production consumer you write. The specific operator chain will change; the shutdown wire-up should not.

public sealed class OrdersConsumer : IHostedService
{
    private readonly ActorSystem _system;
    private readonly ConsumerSettings<string, string> _consumerSettings;
    private readonly CommitterSettings _committerSettings;
    private DrainingControl<Done>? _control;

    public OrdersConsumer(ActorSystem system, ConsumerSettings<string, string> consumerSettings,
        CommitterSettings committerSettings)
    {
        _system = system;
        _consumerSettings = consumerSettings;
        _committerSettings = committerSettings;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        // Materialize exactly once.
        _control = KafkaConsumer
            .CommittableSource(_consumerSettings, Subscriptions.Topics("orders"))
            .SelectAsync(4, async msg =>
            {
                await HandleAsync(msg.Record.Message.Value);
                return (ICommittable)msg.CommitableOffset; // commit only after success
            })
            .ToMaterialized(Committer.Sink(_committerSettings), Keep.Both)
            .MapMaterializedValue(tuple => DrainingControl.Create(tuple.Item1, tuple.Item2))
            .Run(_system.Materializer());
        return Task.CompletedTask;
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        // Stop the source, let in-flight elements finish and commit, then close the consumer.
        if (_control is not null)
            await _control.DrainAndShutdown();
    }

    private static Task HandleAsync(string value) => Task.CompletedTask; // your processing
}

Terminate the actor system only after StopAsync has returned. With Akka.Hosting, register AddAkka before your hosted service: by default the host stops services in reverse registration order, so your drain finishes before the ActorSystem shuts down. Keep the drain shorter than the host's shutdown timeout (HostOptions.ShutdownTimeout, 30 seconds by default) and the pod's grace period.

Where to Go From Here

Three directions build directly on the abstractions introduced here:

• Commit batching: Committer.Sink and Committer.Flow already batch offsets into a CommittableOffsetBatch; CommitterSettings.WithMaxBatch, WithMaxInterval and WithParallelism trade commit traffic against the replay window you traced in the Pitfalls exercise.
• Partitioned sources: CommittablePartitionedSource and PlainPartitionedSource give you one sub-stream per assigned partition, so a slow partition does not hold back the others.
• Transactions (not supported yet): the library contains TransactionalSource and TransactionalFlow/TransactionalSink for Kafka-to-Kafka pipelines, but its README says transactions are not implemented yet (issue #85); the underlying guarantees are covered in "Delivery Semantics & Schema Strategy".

For running these pipelines in production — retries, dead-letter topics, lag dashboards — continue with "Production Patterns & Observability".

✍️ Exercise: sketch the at-least-once graph

Without looking back, write the operator chain for this requirement: consume orders, call an async enrichment API with at most 8 concurrent calls, produce the enriched record to orders-enriched, and commit each input offset only after the enriched record is acknowledged. Include how you would materialize it for graceful shutdown.

DrainingControl<Done> control = KafkaConsumer
    .CommittableSource(consumerSettings, Subscriptions.Topics("orders"))
    .SelectAsync(8, async msg => ProducerMessage.Single(
        new ProducerRecord<string, string>("orders-enriched", msg.Record.Message.Key,
            await EnrichAsync(msg.Record.Message.Value)),
        msg.CommitableOffset))
    .Via(KafkaProducer.FlexiFlow<string, string, ICommittableOffset>(producerSettings))
    .Select(ICommittable (result) => result.PassThrough)
    .ToMaterialized(Committer.Sink(committerSettings), Keep.Both)
    .MapMaterializedValue(tuple => DrainingControl.Create(tuple.Item1, tuple.Item2))
    .Run(materializer);