👀 Views: 93 💬 Answers: 1 📅 Created: 2025-06-29

akka-stream backpressure scala Scala

I'm performance testing and I'm migrating some code and I'm integrating two systems and I've been banging my head against this for hours... I'm working on a Scala 2.13.8 application that uses Akka Streams for processing a large stream of data from Kafka. I’ve set up my stream with backpressure, but I'm noticing that my downstream processing stage is getting overwhelmed and sometimes fails with a `StreamTcp` exception: `Tcp.CommandFailed: Tcp.Connect(...)` when the upstream is producing messages faster than the downstream can handle. Here’s a simplified version of my stream setup: ```scala import akka.actor.ActorSystem import akka.kafka.scaladsl.Consumer import akka.kafka.{ConsumerSettings, Subscriptions} import akka.stream.scaladsl.{Sink, Source} import akka.stream.{ActorMaterializer, OverflowStrategy} import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.common.serialization.StringDeserializer implicit val system = ActorSystem("KafkaStream") implicit val materializer = ActorMaterializer() val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer) .withBootstrapServers("localhost:9092") .withGroupId("my-group") .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") val source = Consumer.plainSource(consumerSettings, Subscriptions.topics("my-topic")) val sink = Sink.foreach[String] { message => // Simulate processing time Thread.sleep(100) } source .buffer(100, OverflowStrategy.backpressure) .to(sink) .run() ``` I’ve tried increasing the buffer size, but it doesn't seem to solve the question. I also tried adjusting the parallelism of the processing stage using `mapAsync`, but I still encounter the same scenario. What's the best way to ensure that my downstream processing can keep up with the incoming stream, or at the very least, handle the backpressure more gracefully? For context: I'm using Scala on Ubuntu. I'm developing on Ubuntu 22.04 with Scala. What's the best practice here?