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

[Nikolay Izhikov (Jira)]

     [ 
https://issues.apache.org/jira/browse/IGNITE-16523?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Nikolay Izhikov updated IGNITE-16523:
-------------------------------------
    Description: 
When WALForceArchiveTimeout set node failed to restart.

{code:java}

/** */
@RunWith(Parameterized.class)
public class RestartWithWalForceArchiveTimeoutTest extends 
GridCommonAbstractTest {
    /** */
    @Parameterized.Parameter
    public WALMode walMode;

    /** {@inheritDoc} */
    @Override protected IgniteConfiguration getConfiguration(String 
igniteInstanceName) throws Exception {
        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);

        cfg.setConsistentId(igniteInstanceName);

        cfg.setDataStorageConfiguration(new DataStorageConfiguration()
            .setWalMode(walMode)
            .setWalForceArchiveTimeout(60 * 60 * 1000) // 1 hour to make sure 
auto archive will not work.
            .setDefaultDataRegionConfiguration(new 
DataRegionConfiguration().setPersistenceEnabled(true)));

        return cfg;
    }

    /** */
    @Parameterized.Parameters(name = "walMode={0}")
    public static Collection<?> parameters() {
        return EnumSet.of(WALMode.FSYNC, WALMode.LOG_ONLY, WALMode.BACKGROUND);
    }

    /** */
    @Test
    public void testRestart() throws Exception {
        stopAllGrids(true);

        cleanPersistenceDir();

        Supplier<IgniteEx> restart = () -> {
            stopAllGrids(true);

            try {
                IgniteEx ign = startGrid(getConfiguration("ignite-0"));

                ign.cluster().state(ACTIVE);

                return ign;
            }
            catch (Exception e) {
                throw new RuntimeException(e);
            }
        };

        IgniteEx ign = restart.get();

        IgniteCache<Integer, AbstractCdcTest.User> cache = 
ign.getOrCreateCache(DEFAULT_CACHE_NAME);

        addData(cache, 0, 100);

        ign = restart.get();

        cache = ign.getOrCreateCache(DEFAULT_CACHE_NAME);

        addData(cache, 100, 200);

        ign = restart.get();

        cache = ign.getOrCreateCache(DEFAULT_CACHE_NAME);

        addData(cache, 200, 300);

        restart.get();
    }
}

{code}

  was:
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}


> 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
>            Priority: Major
>              Labels: IEP-59
>
> When WALForceArchiveTimeout set node failed to restart.
> {code:java}
> /** */
> @RunWith(Parameterized.class)
> public class RestartWithWalForceArchiveTimeoutTest extends 
> GridCommonAbstractTest {
>     /** */
>     @Parameterized.Parameter
>     public WALMode walMode;
>     /** {@inheritDoc} */
>     @Override protected IgniteConfiguration getConfiguration(String 
> igniteInstanceName) throws Exception {
>         IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
>         cfg.setConsistentId(igniteInstanceName);
>         cfg.setDataStorageConfiguration(new DataStorageConfiguration()
>             .setWalMode(walMode)
>             .setWalForceArchiveTimeout(60 * 60 * 1000) // 1 hour to make sure 
> auto archive will not work.
>             .setDefaultDataRegionConfiguration(new 
> DataRegionConfiguration().setPersistenceEnabled(true)));
>         return cfg;
>     }
>     /** */
>     @Parameterized.Parameters(name = "walMode={0}")
>     public static Collection<?> parameters() {
>         return EnumSet.of(WALMode.FSYNC, WALMode.LOG_ONLY, 
> WALMode.BACKGROUND);
>     }
>     /** */
>     @Test
>     public void testRestart() throws Exception {
>         stopAllGrids(true);
>         cleanPersistenceDir();
>         Supplier<IgniteEx> restart = () -> {
>             stopAllGrids(true);
>             try {
>                 IgniteEx ign = startGrid(getConfiguration("ignite-0"));
>                 ign.cluster().state(ACTIVE);
>                 return ign;
>             }
>             catch (Exception e) {
>                 throw new RuntimeException(e);
>             }
>         };
>         IgniteEx ign = restart.get();
>         IgniteCache<Integer, AbstractCdcTest.User> cache = 
> ign.getOrCreateCache(DEFAULT_CACHE_NAME);
>         addData(cache, 0, 100);
>         ign = restart.get();
>         cache = ign.getOrCreateCache(DEFAULT_CACHE_NAME);
>         addData(cache, 100, 200);
>         ign = restart.get();
>         cache = ign.getOrCreateCache(DEFAULT_CACHE_NAME);
>         addData(cache, 200, 300);
>         restart.get();
>     }
> }
> {code}



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