Kafka Streams App exits within seconds of starting

user_12953677 0
Viewed 328

I want my application to keep going and not stop at all but my app exits immediately after few seconds without giving any sort of error or warning. I have been trying to figure out the reason but I haven't found anything useful so far.

I have given my resources configuration file path under build and run option.

I am using config factory to read files from resources json.

I am using scala programming language with idea intelliJ. Kafka in Local server, sql server as source and mongodb as target. kafka tool to view topic and data.

apache-kafka-streams apache-kafka exit scala
2 Answers

In my experience, sudden death with no logging in Kafka Streams (and JVM apps in general) is induced by a native OOM.

Kafka Streams uses RocksDB by default for storing stream state. It’s a bit memory-hungry with default settings. Since RocksDB is native, it won’t be constrained by or use your allocated heap, and if the native library fails to allocate memory, the JVM will die without an exception being thrown.

You can validate this hypothesis by running: dmesg | tail -n 10 Or some such, immediately after the failure. You should see a fairly unambiguous report of a process being sacrificed due to a failed memory allocation.

You can tweak the RocksDB config settings by providing an implementation of RocksDBConfigSetter to stream properties. Reducing your topic partition count in test environments can help a lot, as your number of RocksDB tables will scale with the number of partitions.

user_4368720

Can you specify which kafka-client do you use? https://github.com/zio/zio-kafka works for me out of the box:

import zio._
import zio.console._
import zio.duration._
import zio.kafka.consumer._
import zio.kafka.serde._

object Main extends App {
  def run(args: List[String]) = {
    val settings: ConsumerSettings =
      ConsumerSettings(List("localhost:9092"))
        .withGroupId("group")
        .withCloseTimeout(30.seconds)

    val subscription = Subscription.topics("topic")

    Consumer.consumeWith(settings, subscription, Serde.string, Serde.string) { case (key, value) =>
      putStrLn(s"Received message ${key}: ${value}")
        .ignore
      // Perform an effect with the received message
    }.exitCode
  }

}
user_62977560
Related