feat: add KsqlUncaughtExceptionHandler and new KsqlRestConfig for enabling it

[stevenpyzhang]

Description

Currently, there's no way to bring back a thread that dies due to exception. This PR adds a UncaughtExceptionHandler to the CommandRunner thread which will take down the server when the thread dies due to an unknown exception. the executorService was also changed to execute() because submit() doesn't call UncaughtExceptionHandler.

https://stackoverflow.com/questions/1838923/why-is-uncaughtexceptionhandler-not-called-by-executorservice

Example: if the CommandRunner thread encounters problems when polling the commandTopic, the exception returned isn't being handled currently and the thread will die. The server is then left in a state where it can't process any more commands now that the thread is dead. The only way to restore functionality is to just restart the entire server.

This PR also adds a new KafkaStreamsUncaughtExceptionHandler() for persistent queries since if there's an exception in them, the server can still function normally.

Testing done

Set the new config and threw a RuntimeException from CommandRunner thread and it closed the server after it was thrown.

When the config wasn't set and the exception thrown from CommandRunner, the thread died and the server continued running.

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[rodesai]

I thought the plan was to use https://docs.oracle.com/javase/7/docs/api/java/lang/Thread.html#setDefaultUncaughtExceptionHandler(java.lang.Thread.UncaughtExceptionHandler) to set this globally. Any reason we're not doing that?

[rodesai]

Couple things:

Can we also write the exception to stderr here?

We should call org.apache.log4j.LogManager.shutdown() to ensure that all the buffered logs (including the one above) get flushed. Ideally this would get passed in the constructor as a Runnable:

class KsqlUncaughtExceptionHandler implements UncaughtExceptionHandler {
    private final Runnable flusher;

    public void KsqlUncaughtExceptionHandler(final Runnable flusher) {
        this.flusher = flusher;
    }
}
...
new KsqlUncaughtExceptionHandler(LogManager::flush);

[stevenpyzhang]

I'll add it, but is there a reason why we don't call LogManager.shutdown() currently when the server exits? I don't see any other uses of it.

[rodesai]

We should definitely call it when the server exits cleanly as well. It's more critical here because we might lose the exact information we need to help debug the issue that made the thread crash.

[stevenpyzhang]

I thought the plan was to use https://docs.oracle.com/javase/7/docs/api/java/lang/Thread.html#setDefaultUncaughtExceptionHandler(java.lang.Thread.UncaughtExceptionHandler) to set this globally. Any reason we're not doing that?

@rodesai I had that originally. Could there be other threads that can safely die other than the Stream threads? It seems a bit much to enable this globally by default. Maybe we should add a new config that allows us to turn it on or off (Default is that it's off)?

[rodesai]

Could there be other threads that can safely die other than the Stream threads? It seems a bit much to enable this globally by default. Maybe we should add a new config that allows us to turn it on or off (Default is that it's off)?

I wouldn't ever expect a thread to exit this way. If it does then we're probably in some degraded state. There is some risk of our dependencies having bugs where their threads die, but it would be reasonable to keep running. Let's see how it goes and we can add more targeted handling if we need to.

[rodesai]

LGTM, modulo the comment about config names