Cats-effect: Kafka streams error with IOApp

[barambani]

In a project I'm working on I'm noticing a behaviour that might be expected but that I don't understand completely. I tried to implement a kafka streams application using the IOApp, mainly asynchronously starting the KafkaStreams in the run. At runtime I get

java.lang.ExceptionInInitializerError
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:538)
    at Main$.$anonfun$run$1(Main.scala:22)
    at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:85)
    at cats.effect.internals.IORunLoop$.start(IORunLoop.scala:34)
    at cats.effect.internals.IOBracket$.$anonfun$apply$1(IOBracket.scala:36)
    at cats.effect.internals.IOBracket$.$anonfun$apply$1$adapted(IOBracket.scala:33)
    at cats.effect.internals.IORunLoop$RestartCallback.start(IORunLoop.scala:328)
    at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:117)
    at cats.effect.internals.IORunLoop$RestartCallback.signal(IORunLoop.scala:336)
    at cats.effect.internals.IORunLoop$RestartCallback.apply(IORunLoop.scala:357)
    at cats.effect.internals.IORunLoop$RestartCallback.apply(IORunLoop.scala:303)
    at cats.effect.internals.IOShift$Tick.run(IOShift.scala:36)
    at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.streams.errors.LogAndFailExceptionHandler for configuration default.deserialization.exception.handler: Class org.apache.kafka.streams.errors.LogAndFailExceptionHandler could not be found.
    at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:724)
    at org.apache.kafka.common.config.ConfigDef$ConfigKey.<init>(ConfigDef.java:1043)
    at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:146)
    at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:166)
    at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:205)
    at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:367)
    at org.apache.kafka.common.config.ConfigDef.define(ConfigDef.java:380)
    at org.apache.kafka.streams.StreamsConfig.<clinit>(StreamsConfig.java:505)
    ... 17 more

I have been able to reduce the case a bit in this repo. When I change the app to extend IOApp.WithContext like in this comment it runs ok. It looks like that to work, the kafka stream needs to shift to another context. What could it be the reason for that ? Should be expected or is it an issue ?

[rossabaker]

It can be minimized as:

object Main extends IOApp {
  def run(args: List[String]): IO[ExitCode] =
    IO(throw new VirtualMachineError {})
}

IO.apply does not catch and suspend fatal (i.e., not caught by NonFatal) errors. IOApp is effectively running:

IO(throw new VirtualMachineError {}).start.flatMap(_.join).unsafeRunSync()

The uncaught error is logged, but the fiber returned by start never completes, so the joined IO never completes, so unsafeRunSync() blocks the main thread, so the application never shuts down.

By trying not to handle fatal errors, we've ensured in this case that a fatal error was not actually fatal.

[barambani]

Thanks for the explanation, makes sense. Anyway, I was wondering why the fatal error happens in the first place. I mean the
Class org.apache.kafka.streams.errors.LogAndFailExceptionHandler could not be found
doesn't make any sense to me, as it's in the same Jar as KafkaStreams that's actually using it. The main difference I see is that, when I run the IOApp on another execution context the error doesn't happen at all and the stream starts fine. Anyway I'm happy to close this if the opinion is that the issue here is not in this library but in the Kafka Strams'. Thanks for the help. And thanks for the work across the board here. This has become a really elegant and useful tool.

[rossabaker]

Thanks for the kind words.

I don't have any insight into kafka-streams, but I'm still pondering what we might be able to do about an asynchronous fatal errors and terminating the main fiber to give a better troubleshooting experience for this class of problem.

[barambani]

I see the point, and it will be a further step forward. In relation to my issue I have a workaround, as said above. I will try to understand the root cause for few more days, as I'm now looking also into sbt. Something funny is happening with the class loader when the demon kafka streams thread in start runs in the IOApp run loop. I will close this ticket at the end.

[djspiewak]

@rossabaker I think I may be missing something… Why doesn't IOApp just run unsafeRunSync()? Why does it start and then immediately join a fiber?

[alexandru]

@djspiewak we need to cancel the running process safely when the app gets an interrupt signal (e.g. like when you're pressing Ctrl+C or do a kill).

In order to do that the logic, Ross did it like this:

1. it does a start in order to get a Fiber back
2. with the Fiber we install a shutdown hook (via Runtime.addShutdownHook) that will trigger Fiber#cancel on interruption

Note that the fork is kind of mandatory in such a case, because the shutdown hook needs to be installed after you get a cancellation token and you can only get that cancellation token after you start the execution.

If for example the logic of that IO would execute on the main thread, that thread will be blocked from installing any shutdown hook or to signal the proper cancellation token to the logic used in the shutdown hook. I mean we could use unsafeRunCancelable which executes the IO immediately and gives you a cancellation token. But if the IO executes on the current thread, then when that function returns it's going to be too late for installing the shutdown hook.

Does this make sense?

[alexandru]

Btw, here is the relevant code: https://github.com/typelevel/cats-effect/blob/master/core/jvm/src/main/scala/cats/effect/internals/IOAppPlatform.scala#L43

So it's not joining immediately. It first installs the shutdown hook and then joins.

[djspiewak]

Does this make sense?

Yep! Thanks for the explanation, @alexandru!

[barambani]

Other than the workaround here, please see a proper solution in the linked issue. I'm closing this one.

UPDATE: that fix actually doesn't work as, even if it starts fine, it prevents the fiber from being cancelled properly on interruption (like after a Ctrl-C). In that case there is some finalisation that needs to be done (kafka streams close) that get skipped completely. I still close this as I can live with the workaround above.

[barambani]

In relation to the cancellation issue when forking see also https://github.com/sbt/sbt/issues/2274