public State state() {
        // we do not need to use the stat lock since the variable is volatile
        return state;
    }
@Test
    public void shouldDieOnInvalidOffsetExceptionWhileRunning() throws Exception {
        final StateStore globalStore = builder.globalStateStores().get(GLOBAL_STORE_NAME);
        initializeConsumer();
        startAndSwallowError();

        TestUtils.waitForCondition(
            () -> globalStreamThread.state() == RUNNING,
            10 * 1000,
            "Thread never started.");

        mockConsumer.updateEndOffsets(Collections.singletonMap(topicPartition, 1L));
        mockConsumer.addRecord(record(GLOBAL_STORE_TOPIC_NAME, 0, 0L, "K1".getBytes(), "V1".getBytes()));

        TestUtils.waitForCondition(
            () -> mockConsumer.position(topicPartition) == 1L,
            10 * 1000,
            "Input record never consumed");

        mockConsumer.setPollException(new InvalidOffsetException("Try Again!") {
            @Override
            public Set<TopicPartition> partitions() {
                return Collections.singleton(topicPartition);
            }
        });

        TestUtils.waitForCondition(
            () -> globalStreamThread.state() == DEAD,
            10 * 1000,
            "GlobalStreamThread should have died."
        );
        globalStreamThread.join();

        assertThat(globalStore.isOpen(), is(false));
        assertFalse(new File(baseDirectoryName + File.separator + "testAppId" + File.separator + "global").exists());
    }