@Override
    public CompletableFuture<Void> putStateAsync(String key, ByteBuffer value) {
        ensureStateEnabled();
        return defaultStateStore.putAsync(key, value);
    }
@Test
    public void testPutStateStateEnabled() throws Exception {
        context.defaultStateStore = mock(BKStateStoreImpl.class);
        ByteBuffer buffer = ByteBuffer.wrap("test-value".getBytes(UTF_8));
        context.putStateAsync("test-key", buffer);
        verify(context.defaultStateStore, times(1)).putAsync(eq("test-key"), same(buffer));
    }