Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java (Java)

Code before:
                ),
                Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG))
            );

            try {
                stateConsumer.initialize();
            } catch (final InvalidOffsetException recoverableException) {
                log.error(
                    "Bootstrapping global state failed due to inconsistent local state. Will attempt to clean up the local state. You can restart KafkaStreams to recover from this error.",
                    recoverableException
                );

                closeStateConsumer(stateConsumer, true);

                throw new StreamsException(
                    "Bootstrapping global state failed. You can restart KafkaStreams to recover from this error.",
                    recoverableException
                );
            }

            setState(RUNNING);
            return stateConsumer;
        } catch (final StreamsException fatalException) {
            closeStateConsumer(stateConsumer, false);
            startupException = fatalException;
        } catch (final Throwable throwable) {
            closeStateConsumer(stateConsumer, false);
            startupException = new StreamsException("Exception caught during initialization of GlobalStreamThread", throwable);
        } finally {
            initializationLatch.countDown();
        }
        return null;
    }

    private void closeStateConsumer(final StateConsumer stateConsumer, final boolean wipeStateStore) {
        if (stateConsumer != null) {
            try {
                stateConsumer.close(wipeStateStore);
            } catch (final IOException e) {
                log.error("Failed to close state consumer due to the following error:", e);
            }
        }
    }

    @Override
    public synchronized void start() {
        super.start();
        try {
            initializationLatch.await();
        } catch (final InterruptedException e) {
            currentThread().interrupt();

Review comment (suggestion):
```suggestion
        } catch (final Throwable fatalException) {
```

Code after:
                ),
                Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG))
            );

            try {
                stateConsumer.initialize();
            } catch (final InvalidOffsetException recoverableException) {
                log.error(
                    "Bootstrapping global state failed due to inconsistent local state. Will attempt to clean up the local state. You can restart KafkaStreams to recover from this error.",
                    recoverableException
                );

                closeStateConsumer(stateConsumer, true);

                throw new StreamsException(
                    "Bootstrapping global state failed. You can restart KafkaStreams to recover from this error.",
                    recoverableException
                );
            }

            setState(RUNNING);
            return stateConsumer;
        } catch (final StreamsException fatalException) {
            closeStateConsumer(stateConsumer, false);
            startupException = fatalException;
        } catch (final Throwable fatalException) {
            closeStateConsumer(stateConsumer, false);
            startupException = new StreamsException("Exception caught during initialization of GlobalStreamThread", fatalException);
        } finally {
            initializationLatch.countDown();
        }
        return null;
    }

    private void closeStateConsumer(final StateConsumer stateConsumer, final boolean wipeStateStore) {
        if (stateConsumer != null) {
            try {
                stateConsumer.close(wipeStateStore);
            } catch (final IOException e) {
                log.error("Failed to close state consumer due to the following error:", e);
            }
        }
    }

    @Override
    public synchronized void start() {
        super.start();
        try {
            initializationLatch.await();
        } catch (final InterruptedException e) {
            currentThread().interrupt();