public void shutdown() {
        // one could call shutdown() multiple times, so ignore subsequent calls
        // if already shutting down or dead
        setState(PENDING_SHUTDOWN);
    }
@Test
    public void shouldCloseStateStoresOnClose() throws Exception {
        initializeConsumer();
        startAndSwallowError();
        final StateStore globalStore = builder.globalStateStores().get(GLOBAL_STORE_NAME);
        assertTrue(globalStore.isOpen());
        globalStreamThread.shutdown();
        globalStreamThread.join();
        assertFalse(globalStore.isOpen());
    }