[jira] [Created] (IGNITE-16523) CLONE - All data comited before node stop should become available for CDC consumer

[Nikolay Izhikov (Jira)]

Nikolay Izhikov created IGNITE-16523:
----------------------------------------

             Summary: CLONE - All data comited before node stop should become 
available for CDC consumer
                 Key: IGNITE-16523
                 URL: https://issues.apache.org/jira/browse/IGNITE-16523
             Project: Ignite
          Issue Type: Improvement
            Reporter: Nikolay Izhikov
            Assignee: Nikolay Izhikov


Ignite should make sure that all data committed before node stop are available 
to CDC consumer in case graceful shutdown. Basically, this means Ignite should 
execute the current WAL segment archivation on node stop.

Possible test to check it:
{code:java}

    /** */
    @Test
    @Ignore("Not implemented yet")
    public void testReadAfterNodeStop() throws Exception {
        cleanPersistenceDir();

        AtomicInteger cnt = new AtomicInteger();

        TestCDCConsumer cnsmr = new TestCDCConsumer();

        // Restart node several time to make sure we can continue after 
gracefull shutdown.
        for (int restarts = 0; restarts < 2; restarts++) {
            IgniteConfiguration cfg = getConfiguration("ignite-0");

            Ignite ign = startGrid(cfg);

            ign.cluster().state(ACTIVE);

            long startCnt = cnt.get();

            runAsync(() -> {
                IgniteCache<Integer, User> cache = 
ign.getOrCreateCache(DEFAULT_CACHE_NAME);

                while (true) {
                    int key = cnt.getAndIncrement();

                    try {
                        cache.put(key, createUser(key));
                    }
                    catch (Exception e) {
                        cnt.decrementAndGet();

                        throw e;
                    }
                }
            });

            waitForCondition(() -> cnt.get() - startCnt > KEYS_CNT, 
getTestTimeout());

            ign.close();

            ChangeDataCapture cdc = new ChangeDataCapture(cfg, null, 
cdcConfig(cnsmr));

            IgniteInternalFuture<?> fut = runAsync(cdc);

            assertTrue(waitForSize(cnt.get(), DEFAULT_CACHE_NAME, UPDATE, 
getTestTimeout(), cnsmr));

            fut.cancel();

            List<Integer> keys = cnsmr.keys(UPDATE, 
cacheId(DEFAULT_CACHE_NAME));

            assertTrue(cnt.get() <= keys.size());

            for (int i = 0; i < cnt.get(); i++)
                assertTrue(keys.contains(i));
        }
    }

{code}



--
This message was sent by Atlassian Jira
(v8.20.1#820001)