Support for setting uncaught exception handler

[guillermocalvo]

Closes #424

[guillermocalvo]

SonarCloud Quality Gate failed.   

33.3% Coverage

@sdelamo

SonarQube keeps reporting low coverage, but I added specific tests to cover new code.

• KafkaStreamsFactorySpec
• UncaughtExceptionsSpec

When I run them locally, I get 100% coverage so I'm wondering if there might be something wrong in the project config.

I tried several approaches to try and make SonarQube acknowledge the actual code coverage:

• I made KafkaStreamsFactory.setUncaughtExceptionHandler package-protected instead of private.
• I made KafkaStreamsFactory.setUncaughtExceptionHandler instance method instead of static.
• I made KafkaStreamsFactorySpec start an actual kafka container instead of simply invoking the method.

But sadly, no luck.

My next approach would be to move the new code to its own @Singleton and make KafkaStreamsFactory depend on it, but I think it may be a bit overkill.

Any ideas what I can do to convince SonarQube that the new code is actually covered by the new tests?

[sdelamo]

can we do something like?

    /**
     * Make an uncaught exception handler for a given kafka streams configuration.
     *
     * @param properties The kafka streams configuration.
     * @return An optional exception handler if {@code uncaught-exception-handler} was configured.
     */
    private Optional<StreamsUncaughtExceptionHandler> makeUncaughtExceptionHandler(Properties properties) {
        return Optional.ofNullable(properties.getProperty(UNCAUGHT_EXCEPTION_HANDLER_PROPERTY))
            .filter(not(String::isBlank))
            .flatMap(this::makeUncaughtExceptionHandler);
    }
    
    private Optional<StreamsUncaughtExceptionHandler> makeUncaughtExceptionHandler(String action) {
                try {
                    final StreamThreadExceptionResponse response = StreamThreadExceptionResponse.valueOf(action.toUpperCase());
                    return exception -> {
                        if (LOG.isWarnEnabled()) {
                            LOG.warn("Responding with {} to unexpected exception thrown by kafka stream thread", response, exception);
                        }
                        return Optional.of(response);
                    };
                } catch (IllegalArgumentException e) {
                    if (LOG.isWarnEnabled()) {
                        LOG.warn("Ignoring illegal exception handler: {}. Please use one of: {}", action,
                            asList(StreamThreadExceptionResponse.values()));
                    }
                    return Optional.empty();
                }
            });
    }

[sonarcloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

88.9% Coverage
0.0% Duplication