@Override
    public MetadataStore create(String metadataURL, MetadataStoreConfig metadataStoreConfig,
                                boolean enableSessionWatcher) throws MetadataStoreException {
        return new LocalMemoryMetadataStore(metadataURL, metadataStoreConfig);
    }
@Test
    public void testNotifyEvent() throws Exception {
        TestMetadataEventSynchronizer sync = new TestMetadataEventSynchronizer();
        @Cleanup
        MetadataStore store1 = MetadataStoreFactory.create("memory:local",
                MetadataStoreConfig.builder().synchronizer(sync).build());

        String path = "/test";
        byte[] value = "value".getBytes(StandardCharsets.UTF_8);
        store1.put(path, value, Optional.empty()).join();

        assertTrue(store1.exists(path).join());
        MetadataEvent event = sync.notifiedEvents.get(path);
        Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> event != null);
        assertNotNull(event);
        assertEquals(event.getPath(), path);
        assertEquals(event.getValue(), value);
        assertEquals(event.getOptions(), EMPTY_SET);
        assertEquals(event.getType(), NotificationType.Modified);
        assertEquals(event.getSourceCluster(), sync.clusterName);
        assertNull(event.getExpectedVersion());

        // (2) with expected version
        long exptectedVersion = 0L;
        for (; exptectedVersion < 4; exptectedVersion++) {
            sync.notifiedEvents.remove(path);
            store1.put(path, value, Optional.of(exptectedVersion)).join();
            MetadataEvent event2 = sync.notifiedEvents.get(path);
            Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> event2 != null);
            assertNotNull(event2);
            assertEquals(event2.getPath(), path);
            assertEquals((long) event2.getExpectedVersion(), exptectedVersion);
            assertEquals(event2.getType(), NotificationType.Modified);
        }

        // (3) delete node
        sync.notifiedEvents.remove(path);
        store1.delete(path, Optional.of(exptectedVersion)).join();
        MetadataEvent event2 = sync.notifiedEvents.get(path);
        Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> event2 != null);
        assertNotNull(event2);
        assertEquals(event2.getPath(), path);
        assertEquals((long) event2.getExpectedVersion(), exptectedVersion);
        assertEquals(event2.getType(), NotificationType.Deleted);
        assertEquals(event2.getSourceCluster(), sync.clusterName);
        assertEquals(event2.getOptions(), EMPTY_SET);
    }