@Override
    public MetadataStore create(String metadataURL, MetadataStoreConfig metadataStoreConfig,
                                boolean enableSessionWatcher) throws MetadataStoreException {
        return new LocalMemoryMetadataStore(metadataURL, metadataStoreConfig);
    }
@Test
    public void testIsIgnoreEvent() throws Exception {

        TestMetadataEventSynchronizer sync = new TestMetadataEventSynchronizer();
        @Cleanup
        AbstractMetadataStore store1 = (AbstractMetadataStore) MetadataStoreFactory.create("memory:local",
                MetadataStoreConfig.builder().synchronizer(sync).build());

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

        long time1 = Instant.now().toEpochMilli();
        long time2 = time1 -5;
        Stat stats = new Stat(path, 0, time2, time2, false, false);
        GetResult eixistingData = new GetResult(value1, stats);
        // (1) ignore due to Ephemeral node
        MetadataEvent event = new MetadataEvent(path, value1, Sets.newHashSet(CreateOption.Ephemeral), 0L,
                time1, sync.getClusterName(), NotificationType.Modified);
        assertTrue(store1.shouldIgnoreEvent(event, eixistingData));
        // (2) ignore due to invalid expected version
        event = new MetadataEvent(path, value1, EMPTY_SET, 10L/*invalid-version*/,
                time1, sync.getClusterName(), NotificationType.Modified);
        assertTrue(store1.shouldIgnoreEvent(event, eixistingData));
        // (3) accept with valid conditions
        event = new MetadataEvent(path, value1, EMPTY_SET, 0L,
                time1, sync.getClusterName(), NotificationType.Modified);
        assertFalse(store1.shouldIgnoreEvent(event, eixistingData));
        // (4) Ignore due to invalid cluster name
        event = new MetadataEvent(path, value1, EMPTY_SET, 0L,
                time1, null, NotificationType.Modified);
        assertTrue(store1.shouldIgnoreEvent(event, eixistingData));
        // (5) consider due to same timestamp and correct expected version on the same cluster
        event = new MetadataEvent(path, value1, EMPTY_SET, 0L,
                time2, sync.getClusterName(), NotificationType.Modified);
        assertFalse(store1.shouldIgnoreEvent(event, eixistingData));
        // (6) Ignore due to same timestamp but different expected version on the same cluster
        event = new MetadataEvent(path, value1, EMPTY_SET, 10L,
                time2, sync.getClusterName(), NotificationType.Modified);
        assertTrue(store1.shouldIgnoreEvent(event, eixistingData));
        // (7) consider due to same timestamp but expected version=-1 on the same cluster
        event = new MetadataEvent(path, value1, EMPTY_SET, null,
                time2, sync.getClusterName(), NotificationType.Modified);
        assertFalse(store1.shouldIgnoreEvent(event, eixistingData));
        // (8) Ignore due to less timestamp on the same cluster
        event = new MetadataEvent(path, value1, EMPTY_SET, 0L,
                time2-5, sync.getClusterName(), NotificationType.Modified);
        assertTrue(store1.shouldIgnoreEvent(event, eixistingData));
        // (9) consider "uest" > "test" and same timestamp
        event = new MetadataEvent(path, value1, EMPTY_SET, 0L,
                time2, "uest", NotificationType.Modified);
        assertFalse(store1.shouldIgnoreEvent(event, eixistingData));
        // (10) ignore "uest" > "test" and less timestamp
        event = new MetadataEvent(path, value1, EMPTY_SET, 0L,
                time2-5, "uest", NotificationType.Modified);
        assertTrue(store1.shouldIgnoreEvent(event, eixistingData));
        // (11) ignore "rest" < "test" and same timestamp
        event = new MetadataEvent(path, value1, EMPTY_SET, 0L,
                time2, "rest", NotificationType.Modified);
        assertTrue(store1.shouldIgnoreEvent(event, eixistingData));
    }