modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/cdc/RealtimeCdcBufferTest.java - ignite - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.ignite.internal.processors.cache.persistence.cdc;

 import java.io.File;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.cdc.CdcEvent;
 import org.apache.ignite.cluster.ClusterState;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.DataRegionConfiguration;
 import org.apache.ignite.configuration.DataStorageConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.configuration.WALMode;
 import org.apache.ignite.failure.StopNodeFailureHandler;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.pagemem.wal.WALIterator;
 import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
 import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
 import org.apache.ignite.internal.processors.cache.persistence.wal.reader.IgniteWalIteratorFactory;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiPredicate;
 import org.apache.ignite.testframework.ListeningTestLogger;
 import org.apache.ignite.testframework.LogListener;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.jetbrains.annotations.Nullable;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;

 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.DFLT_STORE_DIR;
 import static org.junit.Assume.assumeFalse;

 /** */
 @RunWith(Parameterized.class)
 public class RealtimeCdcBufferTest extends GridCommonAbstractTest {
     /** */
     private static final String CONSISTENT_ID = "ID";

     /** */
     private static ListeningTestLogger lsnrLog;

     /** */
     private static CountDownLatch stopLatch;

     /** */
     private ByteBufferCdcConsumer consumer;

     /** */
     private boolean cdcEnabled;

     /** */
     private int maxCdcBufSize;

     /** */
     private AtomicInteger commitCnt;

     /** */
     @Parameterized.Parameter()
     public WALMode walMode;

     /** */
     @Parameterized.Parameters(name = "walMode={0}")
     public static List<Object[]> params() {
         return F.asList(
             new Object[] { WALMode.LOG_ONLY },
             new Object[] { WALMode.BACKGROUND },
             new Object[] { WALMode.FSYNC }
         );
     }

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

         consumer = new ByteBufferCdcConsumer(commitCnt);

         cfg.setDataStorageConfiguration(new DataStorageConfiguration()
             .setMaxCdcBufferSize(maxCdcBufSize)
             .setCdcConsumer(consumer)
             .setDefaultDataRegionConfiguration(new DataRegionConfiguration()
                 .setCdcEnabled(cdcEnabled)
                 .setPersistenceEnabled(true)  // TODO: test for in-memory mode.
             )
             .setWalMode(walMode)
         );

         cfg.setCacheConfiguration(new CacheConfiguration<>(DEFAULT_CACHE_NAME));

         cfg.setFailureHandler(new StopNodeFailureHandler());

         cfg.setGridLogger(lsnrLog);

         cfg.setConsistentId(CONSISTENT_ID);

         return cfg;
     }

     /** */
     @Override protected void beforeTest() throws Exception {
         super.beforeTest();

         lsnrLog = new ListeningTestLogger(log);

         cleanPersistenceDir();

         cdcEnabled = true;

         stopLatch = null;
         commitCnt = null;
         consumer = null;
     }

     /** */
     @Test
     public void testCdcBufferOverflow() throws Exception {
         // TODO: Looks like there is a bug in the FSYNC case: WAL misses some records (HEADER_RECORD).
         assumeFalse(walMode == WALMode.FSYNC);

         maxCdcBufSize = 100 * (int)U.KB;

         checkCdcBufferOverflow(buildEntries(100, 10 * (int)U.KB), true);
     }

     /** */
     @Test
     public void testCdcDisabled() throws Exception {
         cdcEnabled = false;

         checkCdcBufferOverflow(buildEntries(100, 10 * (int)U.KB), false);
     }

     /** */
     @Test
     public void testCdcRecords() throws Exception {
         // TODO: Looks like there is a bug in the FSYNC case: WAL misses some records (HEADER_RECORD).
         assumeFalse(walMode == WALMode.FSYNC);

         maxCdcBufSize = 100 * (int)U.MB;
         commitCnt = new AtomicInteger();

         IgniteEx crd = startGrid(0);

         crd.cluster().state(ClusterState.ACTIVE);

         IgniteCache<Integer, Integer> cache = crd.cache(DEFAULT_CACHE_NAME);

         // Await while cluster is fully activated.
         cache.put(0, 0);

         int expCommitCnt = 5;

         commitCnt.set(expCommitCnt);

         while (commitCnt.get() >= 0)
             cache.put(0, 0);

         forceCheckpoint(crd);

         stopGrid(0);

         try (WALIterator walIt = walIter(null)) {
             int cdcRecCnt = 0;

             while (walIt.hasNext()) {
                 if (walIt.next().getValue().type() == WALRecord.RecordType.REALTIME_CDC_RECORD)
                     cdcRecCnt++;
             }

             assertEquals(expCommitCnt, cdcRecCnt);
         }
     }

     /** */
     @Test
     public void testCdcBufferContent() throws Exception {
         // TODO: Looks like there is a bug in the FSYNC case: WAL misses some records (HEADER_RECORD).
         assumeFalse(walMode == WALMode.FSYNC);

         maxCdcBufSize = 10 * (int)U.MB;

         int entriesCnt = 1;

         List<T2<Integer, byte[]>> expEntries = buildEntries((int)U.KB, entriesCnt);

         checkCdcBufferOverflow(expEntries, false);

         assertEquals(entriesCnt, expEntries.size());

         List<T2<Integer, byte[]>> cdcEntries = consumer.buf.stream()
             .map(e -> new T2<>((int)e.key(), (byte[])e.value()))
             .collect(Collectors.toList());

         assertEquals(expEntries.size(), cdcEntries.size());

         for (int i = 0; i < expEntries.size(); i++) {
             assertEquals(expEntries.get(i).getKey(), cdcEntries.get(i).getKey());
             assertTrue(Arrays.equals(expEntries.get(i).getValue(), cdcEntries.get(i).getValue()));
         }
     }

     /** */
     private List<T2<Integer, byte[]>> buildEntries(int size, int cnt) {
         List<T2<Integer, byte[]>> entries = new ArrayList<>();

         for (int i = 0; i < cnt; i++) {
             byte[] data = new byte[size];

             ThreadLocalRandom.current().nextBytes(data);

             entries.add(new T2<>(i, data));
         }

         return entries;
     }

     /** */
     private void checkCdcBufferOverflow(List<T2<Integer, byte[]>> entries, boolean shouldOverflow) throws Exception {
         stopLatch = cdcEnabled ? new CountDownLatch(entries.size()) : null;

         LogListener lsnr = LogListener.matches("CDC buffer has overflowed. Stop realtime mode of CDC.")
             .times(shouldOverflow ? 1 : 0)
             .build();

         lsnrLog.registerListener(lsnr);

         IgniteEx crd = startGrid(0);

         crd.cluster().state(ClusterState.ACTIVE);

         IgniteCache<Integer, byte[]> cache = crd.cache(DEFAULT_CACHE_NAME);

         for (T2<Integer, byte[]> e: entries)
             cache.put(e.getKey(), e.getValue());

         forceCheckpoint(crd);

         if (stopLatch != null && !shouldOverflow)
             U.awaitQuiet(stopLatch);

         stopGrid(0);

         assertTrue(lsnr.check());

         try (WALIterator walIt = walIter((recType, recPtr) -> recType == WALRecord.RecordType.REALTIME_STOP_CDC_RECORD)) {
             AtomicInteger stopRecCnt = new AtomicInteger();

             walIt.forEach((rec) -> stopRecCnt.incrementAndGet());

             assertEquals(shouldOverflow ? 1 : 0, stopRecCnt.get());
         }
     }

     /** */
     private File walSegments() throws IgniteCheckedException {
         return U.resolveWorkDirectory(
             U.defaultWorkDirectory(),
             DFLT_STORE_DIR + "/wal/" + CONSISTENT_ID, false);
     }

     /** Get iterator over WAL. */
     private WALIterator walIter(@Nullable IgniteBiPredicate<WALRecord.RecordType, WALPointer> filter) throws Exception {
         IgniteWalIteratorFactory factory = new IgniteWalIteratorFactory(log);

         IgniteWalIteratorFactory.IteratorParametersBuilder params = new IgniteWalIteratorFactory.IteratorParametersBuilder()
             .filesOrDirs(walSegments())
             .filter(filter);

         return factory.iterator(params);
     }

     /** */
     private static class ByteBufferCdcConsumer implements CdcBufferConsumer {
         /** */
         private final List<CdcEvent> buf = new ArrayList<>();

         /** */
         private final AtomicInteger commitCnt;

         /** */
         ByteBufferCdcConsumer(@Nullable AtomicInteger commitCnt) {
             this.commitCnt = commitCnt;
         }

         /** {@inheritDoc} */
         @Override public boolean consume(Collection<CdcEvent> data) {
             buf.addAll(data);

             if (stopLatch != null) {
                 for (int i = 0; i < data.size(); i++)
                     stopLatch.countDown();
             }

             return commitCnt != null && commitCnt.decrementAndGet() >= 0;
         }

         /** */
         @Override public void close() {
             for (int i = 0; i < stopLatch.getCount(); i++)
                 stopLatch.countDown();
         }
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.ignite.internal.processors.cache.persistence.cdc;

	import java.io.File;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.List;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.ThreadLocalRandom;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.stream.Collectors;
	import org.apache.ignite.IgniteCache;
	import org.apache.ignite.IgniteCheckedException;
	import org.apache.ignite.cdc.CdcEvent;
	import org.apache.ignite.cluster.ClusterState;
	import org.apache.ignite.configuration.CacheConfiguration;
	import org.apache.ignite.configuration.DataRegionConfiguration;
	import org.apache.ignite.configuration.DataStorageConfiguration;
	import org.apache.ignite.configuration.IgniteConfiguration;
	import org.apache.ignite.configuration.WALMode;
	import org.apache.ignite.failure.StopNodeFailureHandler;
	import org.apache.ignite.internal.IgniteEx;
	import org.apache.ignite.internal.pagemem.wal.WALIterator;
	import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
	import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
	import org.apache.ignite.internal.processors.cache.persistence.wal.reader.IgniteWalIteratorFactory;
	import org.apache.ignite.internal.util.typedef.F;
	import org.apache.ignite.internal.util.typedef.T2;
	import org.apache.ignite.internal.util.typedef.internal.U;
	import org.apache.ignite.lang.IgniteBiPredicate;
	import org.apache.ignite.testframework.ListeningTestLogger;
	import org.apache.ignite.testframework.LogListener;
	import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
	import org.jetbrains.annotations.Nullable;
	import org.junit.Test;
	import org.junit.runner.RunWith;
	import org.junit.runners.Parameterized;

	import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.DFLT_STORE_DIR;
	import static org.junit.Assume.assumeFalse;

	/** */
	@RunWith(Parameterized.class)
	public class RealtimeCdcBufferTest extends GridCommonAbstractTest {
	/** */
	private static final String CONSISTENT_ID = "ID";

	/** */
	private static ListeningTestLogger lsnrLog;

	/** */
	private static CountDownLatch stopLatch;

	/** */
	private ByteBufferCdcConsumer consumer;

	/** */
	private boolean cdcEnabled;

	/** */
	private int maxCdcBufSize;

	/** */
	private AtomicInteger commitCnt;

	/** */
	@Parameterized.Parameter()
	public WALMode walMode;

	/** */
	@Parameterized.Parameters(name = "walMode={0}")
	public static List<Object[]> params() {
	return F.asList(
	new Object[] { WALMode.LOG_ONLY },
	new Object[] { WALMode.BACKGROUND },
	new Object[] { WALMode.FSYNC }
	);
	}

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

	consumer = new ByteBufferCdcConsumer(commitCnt);

	cfg.setDataStorageConfiguration(new DataStorageConfiguration()
	.setMaxCdcBufferSize(maxCdcBufSize)
	.setCdcConsumer(consumer)
	.setDefaultDataRegionConfiguration(new DataRegionConfiguration()
	.setCdcEnabled(cdcEnabled)
	.setPersistenceEnabled(true) // TODO: test for in-memory mode.
	)
	.setWalMode(walMode)
	);

	cfg.setCacheConfiguration(new CacheConfiguration<>(DEFAULT_CACHE_NAME));

	cfg.setFailureHandler(new StopNodeFailureHandler());

	cfg.setGridLogger(lsnrLog);

	cfg.setConsistentId(CONSISTENT_ID);

	return cfg;
	}

	/** */
	@Override protected void beforeTest() throws Exception {
	super.beforeTest();

	lsnrLog = new ListeningTestLogger(log);

	cleanPersistenceDir();

	cdcEnabled = true;

	stopLatch = null;
	commitCnt = null;
	consumer = null;
	}

	/** */
	@Test
	public void testCdcBufferOverflow() throws Exception {
	// TODO: Looks like there is a bug in the FSYNC case: WAL misses some records (HEADER_RECORD).
	assumeFalse(walMode == WALMode.FSYNC);

	maxCdcBufSize = 100 * (int)U.KB;

	checkCdcBufferOverflow(buildEntries(100, 10 * (int)U.KB), true);
	}

	/** */
	@Test
	public void testCdcDisabled() throws Exception {
	cdcEnabled = false;

	checkCdcBufferOverflow(buildEntries(100, 10 * (int)U.KB), false);
	}

	/** */
	@Test
	public void testCdcRecords() throws Exception {
	// TODO: Looks like there is a bug in the FSYNC case: WAL misses some records (HEADER_RECORD).
	assumeFalse(walMode == WALMode.FSYNC);

	maxCdcBufSize = 100 * (int)U.MB;
	commitCnt = new AtomicInteger();

	IgniteEx crd = startGrid(0);

	crd.cluster().state(ClusterState.ACTIVE);

	IgniteCache<Integer, Integer> cache = crd.cache(DEFAULT_CACHE_NAME);

	// Await while cluster is fully activated.
	cache.put(0, 0);

	int expCommitCnt = 5;

	commitCnt.set(expCommitCnt);

	while (commitCnt.get() >= 0)
	cache.put(0, 0);

	forceCheckpoint(crd);

	stopGrid(0);

	try (WALIterator walIt = walIter(null)) {
	int cdcRecCnt = 0;

	while (walIt.hasNext()) {
	if (walIt.next().getValue().type() == WALRecord.RecordType.REALTIME_CDC_RECORD)
	cdcRecCnt++;
	}

	assertEquals(expCommitCnt, cdcRecCnt);
	}
	}

	/** */
	@Test
	public void testCdcBufferContent() throws Exception {
	// TODO: Looks like there is a bug in the FSYNC case: WAL misses some records (HEADER_RECORD).
	assumeFalse(walMode == WALMode.FSYNC);

	maxCdcBufSize = 10 * (int)U.MB;

	int entriesCnt = 1;

	List<T2<Integer, byte[]>> expEntries = buildEntries((int)U.KB, entriesCnt);

	checkCdcBufferOverflow(expEntries, false);

	assertEquals(entriesCnt, expEntries.size());

	List<T2<Integer, byte[]>> cdcEntries = consumer.buf.stream()
	.map(e -> new T2<>((int)e.key(), (byte[])e.value()))
	.collect(Collectors.toList());

	assertEquals(expEntries.size(), cdcEntries.size());

	for (int i = 0; i < expEntries.size(); i++) {
	assertEquals(expEntries.get(i).getKey(), cdcEntries.get(i).getKey());
	assertTrue(Arrays.equals(expEntries.get(i).getValue(), cdcEntries.get(i).getValue()));
	}
	}

	/** */
	private List<T2<Integer, byte[]>> buildEntries(int size, int cnt) {
	List<T2<Integer, byte[]>> entries = new ArrayList<>();

	for (int i = 0; i < cnt; i++) {
	byte[] data = new byte[size];

	ThreadLocalRandom.current().nextBytes(data);

	entries.add(new T2<>(i, data));
	}

	return entries;
	}

	/** */
	private void checkCdcBufferOverflow(List<T2<Integer, byte[]>> entries, boolean shouldOverflow) throws Exception {
	stopLatch = cdcEnabled ? new CountDownLatch(entries.size()) : null;

	LogListener lsnr = LogListener.matches("CDC buffer has overflowed. Stop realtime mode of CDC.")
	.times(shouldOverflow ? 1 : 0)
	.build();

	lsnrLog.registerListener(lsnr);

	IgniteEx crd = startGrid(0);

	crd.cluster().state(ClusterState.ACTIVE);

	IgniteCache<Integer, byte[]> cache = crd.cache(DEFAULT_CACHE_NAME);

	for (T2<Integer, byte[]> e: entries)
	cache.put(e.getKey(), e.getValue());

	forceCheckpoint(crd);

	if (stopLatch != null && !shouldOverflow)
	U.awaitQuiet(stopLatch);

	stopGrid(0);

	assertTrue(lsnr.check());

	try (WALIterator walIt = walIter((recType, recPtr) -> recType == WALRecord.RecordType.REALTIME_STOP_CDC_RECORD)) {
	AtomicInteger stopRecCnt = new AtomicInteger();

	walIt.forEach((rec) -> stopRecCnt.incrementAndGet());

	assertEquals(shouldOverflow ? 1 : 0, stopRecCnt.get());
	}
	}

	/** */
	private File walSegments() throws IgniteCheckedException {
	return U.resolveWorkDirectory(
	U.defaultWorkDirectory(),
	DFLT_STORE_DIR + "/wal/" + CONSISTENT_ID, false);
	}

	/** Get iterator over WAL. */
	private WALIterator walIter(@Nullable IgniteBiPredicate<WALRecord.RecordType, WALPointer> filter) throws Exception {
	IgniteWalIteratorFactory factory = new IgniteWalIteratorFactory(log);

	IgniteWalIteratorFactory.IteratorParametersBuilder params = new IgniteWalIteratorFactory.IteratorParametersBuilder()
	.filesOrDirs(walSegments())
	.filter(filter);

	return factory.iterator(params);
	}

	/** */
	private static class ByteBufferCdcConsumer implements CdcBufferConsumer {
	/** */
	private final List<CdcEvent> buf = new ArrayList<>();

	/** */
	private final AtomicInteger commitCnt;

	/** */
	ByteBufferCdcConsumer(@Nullable AtomicInteger commitCnt) {
	this.commitCnt = commitCnt;
	}

	/** {@inheritDoc} */
	@Override public boolean consume(Collection<CdcEvent> data) {
	buf.addAll(data);

	if (stopLatch != null) {
	for (int i = 0; i < data.size(); i++)
	stopLatch.countDown();
	}

	return commitCnt != null && commitCnt.decrementAndGet() >= 0;
	}

	/** */
	@Override public void close() {
	for (int i = 0; i < stopLatch.getCount(); i++)
	stopLatch.countDown();
	}
	}
	}