You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by na...@apache.org on 2022/03/12 09:31:31 UTC
[ignite] branch master updated: IGNITE-14829 Save DataEntry index inside CDC state (#9828)
This is an automated email from the ASF dual-hosted git repository.
namelchev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 5555e8b IGNITE-14829 Save DataEntry index inside CDC state (#9828)
5555e8b is described below
commit 5555e8b5ee36006325eb151f90669fa7a44aee7c
Author: Nikita Amelchev <ns...@gmail.com>
AuthorDate: Sat Mar 12 12:27:34 2022 +0300
IGNITE-14829 Save DataEntry index inside CDC state (#9828)
---
.../ignite/internal/cdc/CdcConsumerState.java | 25 +--
.../org/apache/ignite/internal/cdc/CdcMain.java | 46 +++---
.../ignite/internal/cdc/WalRecordsConsumer.java | 171 ++++++++++++++-------
.../org/apache/ignite/cdc/AbstractCdcTest.java | 14 +-
.../java/org/apache/ignite/cdc/CdcSelfTest.java | 120 ++++++++++++++-
5 files changed, 287 insertions(+), 89 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java
index 0ac8b57..a9ec369 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java
@@ -27,7 +27,10 @@ import java.util.Iterator;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cdc.CdcConsumer;
import org.apache.ignite.cdc.CdcEvent;
+import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
+import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
+import org.apache.ignite.internal.util.typedef.T2;
import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
@@ -39,7 +42,7 @@ import static org.apache.ignite.internal.processors.cache.persistence.wal.WALPoi
* Change Data Capture Consumer state.
*
* Each time {@link CdcConsumer#onEvents(Iterator)} returns {@code true}
- * current offset in WAL segment saved to file.
+ * current offset in WAL segment and {@link DataEntry} index inside {@link DataRecord} saved to file.
* This allows to the {@link CdcConsumer} to continue consumption of the {@link CdcEvent}
* from the last saved offset in case of fail or restart.
*
@@ -66,14 +69,15 @@ public class CdcConsumerState {
/**
* Saves state to file.
- * @param ptr WAL pointer.
+ *
+ * @param state WAL pointer and index of {@link DataEntry} inside {@link DataRecord}.
*/
- public void save(WALPointer ptr) throws IOException {
+ public void save(T2<WALPointer, Integer> state) throws IOException {
ByteBuffer buf = ByteBuffer.allocate(POINTER_SIZE);
- buf.putLong(ptr.index());
- buf.putInt(ptr.fileOffset());
- buf.putInt(ptr.length());
+ buf.putLong(state.get1().index());
+ buf.putInt(state.get1().fileOffset());
+ buf.putInt(state.get2());
buf.flip();
try (FileChannel ch = FileChannel.open(tmp, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
@@ -82,14 +86,15 @@ public class CdcConsumerState {
ch.force(true);
}
- Files.move(tmp, state, ATOMIC_MOVE, REPLACE_EXISTING);
+ Files.move(tmp, this.state, ATOMIC_MOVE, REPLACE_EXISTING);
}
/**
* Loads CDC state from file.
+ *
* @return Saved state.
*/
- public WALPointer load() {
+ public T2<WALPointer, Integer> load() {
if (!Files.exists(state))
return null;
@@ -102,9 +107,9 @@ public class CdcConsumerState {
long idx = buf.getLong();
int offset = buf.getInt();
- int length = buf.getInt();
+ int entryIdx = buf.getInt();
- return new WALPointer(idx, offset, length);
+ return new T2<>(new WALPointer(idx, offset, 0), entryIdx);
}
catch (IOException e) {
throw new IgniteException("Failed to read state [file=" + state + ']', e);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
index 1b6addc..84ba5cf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
@@ -41,8 +41,7 @@ import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.GridLoggerProxy;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.MarshallerContextImpl;
-import org.apache.ignite.internal.pagemem.wal.WALIterator;
-import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
+import org.apache.ignite.internal.cdc.WalRecordsConsumer.DataEntryIterator;
import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderResolver;
import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings;
@@ -53,6 +52,7 @@ import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
import org.apache.ignite.internal.processors.resource.GridSpringResourceContext;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.startup.cmdline.CdcCommandLineStartup;
@@ -182,7 +182,7 @@ public class CdcMain implements Runnable {
private CdcConsumerState state;
/** Save state to start from. */
- private WALPointer initState;
+ private T2<WALPointer, Integer> initState;
/** Stopped flag. */
private volatile boolean stopped;
@@ -272,11 +272,11 @@ public class CdcMain implements Runnable {
initState = state.load();
if (initState != null) {
- committedSegmentIdx.value(initState.index());
- committedSegmentOffset.value(initState.fileOffset());
+ committedSegmentIdx.value(initState.get1().index());
+ committedSegmentOffset.value(initState.get1().fileOffset());
if (log.isInfoEnabled())
- log.info("Initial state loaded [state=" + initState + ']');
+ log.info("Initial state loaded [ptr=" + initState.get1() + ", idx=" + initState.get2() + ']');
}
consumer.start(mreg, kctx.metric().registry(metricName("cdc", "consumer")));
@@ -438,15 +438,15 @@ public class CdcMain implements Runnable {
curSegmentIdx.value(segmentIdx);
if (initState != null) {
- if (segmentIdx > initState.index()) {
+ if (segmentIdx > initState.get1().index()) {
throw new IgniteException("Found segment greater then saved state. Some events are missed. Exiting! " +
"[state=" + initState + ", segment=" + segmentIdx + ']');
}
- if (segmentIdx < initState.index()) {
+ if (segmentIdx < initState.get1().index()) {
if (log.isInfoEnabled()) {
log.info("Already processed segment found. Skipping and deleting the file [segment=" +
- segmentIdx + ", state=" + initState.index() + ']');
+ segmentIdx + ", state=" + initState.get1().index() + ']');
}
// WAL segment is a hard link to a segment file in the special Change Data Capture folder.
@@ -461,31 +461,33 @@ public class CdcMain implements Runnable {
}
}
- builder.from(initState);
-
- initState = null;
+ builder.from(initState.get1());
}
- try (WALIterator it = factory.iterator(builder)) {
- boolean interrupted = Thread.interrupted();
+ try (DataEntryIterator iter = new DataEntryIterator(factory.iterator(builder))) {
+ if (initState != null) {
+ iter.init(initState.get2());
- while (it.hasNext() && !interrupted) {
- Iterator<DataRecord> iter = F.iterator(it.iterator(), t -> (DataRecord)t.get2(), true);
+ initState = null;
+ }
+
+ boolean interrupted = Thread.interrupted();
+ while (iter.hasNext() && !interrupted) {
boolean commit = consumer.onRecords(iter);
if (commit) {
- assert it.lastRead().isPresent();
+ T2<WALPointer, Integer> curState = iter.state();
- WALPointer ptr = it.lastRead().get().next();
+ assert curState != null;
if (log.isDebugEnabled())
- log.debug("Saving state [ptr=" + ptr + ']');
+ log.debug("Saving state [curState=" + curState + ']');
- state.save(ptr);
+ state.save(curState);
- committedSegmentIdx.value(ptr.index());
- committedSegmentOffset.value(ptr.fileOffset());
+ committedSegmentIdx.value(curState.get1().index());
+ committedSegmentOffset.value(curState.get1().fileOffset());
// Can delete after new file state save.
if (!processedSegments.isEmpty()) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java
index 66853a9..8f87d55 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/WalRecordsConsumer.java
@@ -21,17 +21,24 @@ import java.util.EnumSet;
import java.util.Iterator;
import java.util.NoSuchElementException;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cdc.CdcConsumer;
import org.apache.ignite.cdc.CdcEvent;
+import org.apache.ignite.internal.pagemem.wal.WALIterator;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
import org.apache.ignite.internal.pagemem.wal.record.UnwrappedDataEntry;
+import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
+import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteBiTuple;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgnitePredicate;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.CREATE;
@@ -79,6 +86,20 @@ public class WalRecordsConsumer<K, V> {
return OPERATIONS_TYPES.contains(e.op());
};
+ /** Event transformer. */
+ private static final IgniteClosure<DataEntry, CdcEvent> CDC_EVENT_TRANSFORMER = e -> {
+ UnwrappedDataEntry ue = (UnwrappedDataEntry)e;
+
+ return new CdcEventImpl(
+ ue.unwrappedKey(),
+ ue.unwrappedValue(),
+ (e.flags() & DataEntry.PRIMARY_FLAG) != 0,
+ e.partitionId(),
+ e.writeVersion(),
+ e.cacheId()
+ );
+ };
+
/**
* @param consumer User provided CDC consumer.
* @param log Logger.
@@ -89,70 +110,29 @@ public class WalRecordsConsumer<K, V> {
}
/**
- * Handles record from the WAL.
- * If this method return {@code true} then current offset in WAL will be stored and WAL iteration will be
- * started from it on CDC application fail/restart.
+ * Handles data entries.
+ * If this method return {@code true} then current offset in WAL and {@link DataEntry} index inside
+ * {@link DataRecord} will be stored and WAL iteration will be started from it on CDC application fail/restart.
*
- * @param recs WAL records iterator.
+ * @param entries Data entries iterator.
* @return {@code True} if current offset in WAL should be commited.
*/
- public boolean onRecords(Iterator<DataRecord> recs) {
- Iterator<CdcEvent> evts = new Iterator<CdcEvent>() {
- /** */
- private Iterator<CdcEvent> entries;
-
+ public boolean onRecords(Iterator<DataEntry> entries) {
+ Iterator<CdcEvent> evts = F.iterator(new Iterator<DataEntry>() {
@Override public boolean hasNext() {
- advance();
-
- return hasCurrent();
+ return entries.hasNext();
}
- @Override public CdcEvent next() {
- advance();
-
- if (!hasCurrent())
- throw new NoSuchElementException();
+ @Override public DataEntry next() {
+ DataEntry next = entries.next();
evtsCnt.increment();
lastEvtTs.value(System.currentTimeMillis());
- return entries.next();
- }
-
- private void advance() {
- if (hasCurrent())
- return;
-
- while (recs.hasNext()) {
- entries =
- F.iterator(recs.next().writeEntries().iterator(), this::transform, true, OPERATIONS_FILTER);
-
- if (entries.hasNext())
- break;
-
- entries = null;
- }
+ return next;
}
-
- private boolean hasCurrent() {
- return entries != null && entries.hasNext();
- }
-
- /** */
- private CdcEvent transform(DataEntry e) {
- UnwrappedDataEntry ue = (UnwrappedDataEntry)e;
-
- return new CdcEventImpl(
- ue.unwrappedKey(),
- ue.unwrappedValue(),
- (e.flags() & DataEntry.PRIMARY_FLAG) != 0,
- e.partitionId(),
- e.writeVersion(),
- e.cacheId()
- );
- }
- };
+ }, CDC_EVENT_TRANSFORMER, true, OPERATIONS_FILTER);
return consumer.onEvents(evts);
}
@@ -194,4 +174,91 @@ public class WalRecordsConsumer<K, V> {
@Override public String toString() {
return S.toString(WalRecordsConsumer.class, this);
}
+
+ /** Iterator over {@link DataEntry}. */
+ public static class DataEntryIterator implements Iterator<DataEntry>, AutoCloseable {
+ /** WAL iterator. */
+ private final WALIterator walIter;
+
+ /** Current preloaded WAL record. */
+ private IgniteBiTuple<WALPointer, WALRecord> curRec;
+
+ /** */
+ private DataEntry next;
+
+ /** Index of {@link #next} inside WAL record. */
+ private int entryIdx;
+
+ /** @param walIter WAL iterator. */
+ DataEntryIterator(WALIterator walIter) {
+ this.walIter = walIter;
+
+ advance();
+ }
+
+ /** @return Current state. */
+ T2<WALPointer, Integer> state() {
+ return hasNext() ?
+ new T2<>(curRec.get1(), entryIdx) :
+ new T2<>(curRec.get1().next(), 0);
+ }
+
+ /** Initialize state. */
+ void init(int idx) {
+ for (int i = 0; i < idx; i++) {
+ if (!hasNext())
+ throw new IgniteException("Failed to restore entry index [idx=" + idx + ", rec=" + curRec + ']');
+
+ next();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasNext() {
+ return next != null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public DataEntry next() {
+ if (!hasNext())
+ throw new NoSuchElementException();
+
+ DataEntry e = next;
+
+ next = null;
+
+ advance();
+
+ return e;
+ }
+
+ /** */
+ private void advance() {
+ if (curRec != null) {
+ entryIdx++;
+
+ DataRecord rec = (DataRecord)curRec.get2();
+
+ if (entryIdx < rec.entryCount()) {
+ next = rec.get(entryIdx);
+
+ return;
+ }
+
+ entryIdx = 0;
+ }
+
+ if (!walIter.hasNext())
+ return;
+
+ curRec = walIter.next();
+
+ next = ((DataRecord)curRec.get2()).get(entryIdx);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() throws IgniteCheckedException {
+ walIter.close();
+ }
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java b/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java
index ec43d9e..19bf6ee 100644
--- a/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cdc/AbstractCdcTest.java
@@ -42,6 +42,7 @@ import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.CI3;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.spi.metric.LongMetric;
import org.apache.ignite.spi.metric.MetricExporterSpi;
@@ -86,6 +87,15 @@ public abstract class AbstractCdcTest extends GridCommonAbstractTest {
super.beforeTest();
}
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+
+ cleanPersistenceDir();
+ }
+
/** */
protected CdcMain createCdc(CdcConsumer cnsmr, IgniteConfiguration cfg) {
return createCdc(cnsmr, cfg, null);
@@ -107,8 +117,8 @@ public abstract class AbstractCdcTest extends GridCommonAbstractTest {
return new CdcMain(cfg, null, cdcCfg) {
@Override protected CdcConsumerState createState(Path stateDir) {
return new CdcConsumerState(stateDir) {
- @Override public void save(WALPointer ptr) throws IOException {
- super.save(ptr);
+ @Override public void save(T2<WALPointer, Integer> state) throws IOException {
+ super.save(state);
if (!F.isEmpty(conditions)) {
for (GridAbsPredicate p : conditions) {
diff --git a/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java
index 271e847..04e831c 100644
--- a/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cdc/CdcSelfTest.java
@@ -17,19 +17,25 @@
package org.apache.ignite.cdc;
+import java.io.File;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
@@ -38,8 +44,15 @@ import org.apache.ignite.configuration.WALMode;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cdc.CdcMain;
+import org.apache.ignite.internal.pagemem.wal.WALIterator;
+import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
+import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
+import org.apache.ignite.internal.processors.cache.persistence.wal.reader.IgniteWalIteratorFactory;
+import org.apache.ignite.internal.processors.cache.persistence.wal.reader.IgniteWalIteratorFactory.IteratorParametersBuilder;
+import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.spi.metric.MetricExporterSpi;
import org.apache.ignite.spi.metric.jmx.JmxMetricExporterSpi;
import org.apache.ignite.testframework.junits.WithSystemProperty;
@@ -49,12 +62,15 @@ import org.junit.runners.Parameterized;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_STORAGE_FOLDER_BY_CONSISTENT_ID;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cdc.AbstractCdcTest.ChangeEventType.DELETE;
import static org.apache.ignite.cdc.AbstractCdcTest.ChangeEventType.UPDATE;
import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.configuration.DataStorageConfiguration.DFLT_WAL_ARCHIVE_PATH;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.cacheId;
import static org.apache.ignite.testframework.GridTestUtils.runAsync;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static org.junit.Assume.assumeTrue;
/** */
@RunWith(Parameterized.class)
@@ -108,13 +124,16 @@ public class CdcSelfTest extends AbstractCdcTest {
.setCdcEnabled(true)
.setWalMode(walMode)
.setWalForceArchiveTimeout(WAL_ARCHIVE_TIMEOUT)
- .setDefaultDataRegionConfiguration(new DataRegionConfiguration().setPersistenceEnabled(true)));
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration().setPersistenceEnabled(true))
+ .setWalArchivePath(DFLT_WAL_ARCHIVE_PATH + "/" + U.maskForFileName(igniteInstanceName)));
if (pageSz != 0)
cfg.getDataStorageConfiguration().setPageSize(pageSz);
cfg.setCacheConfiguration(
- new CacheConfiguration<>(TX_CACHE_NAME).setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ new CacheConfiguration<>(TX_CACHE_NAME)
+ .setAtomicityMode(TRANSACTIONAL)
+ .setBackups(1)
);
return cfg;
@@ -188,6 +207,101 @@ public class CdcSelfTest extends AbstractCdcTest {
/** */
@Test
+ public void testReadOneByOneForBackup() throws Exception {
+ assumeTrue("CDC with 2 local nodes can't determine correct PDS directory without specificConsistentId.",
+ specificConsistentId);
+
+ IgniteEx ign = startGrids(2);
+
+ ign.cluster().state(ACTIVE);
+
+ IgniteCache<Integer, Integer> txCache = ign.cache(TX_CACHE_NAME);
+
+ int keysCnt = 3;
+
+ Map<Integer, Integer> batch = primaryKeys(txCache, keysCnt).stream()
+ .collect(Collectors.toMap(key -> key, val -> val, (a, b) -> a, TreeMap::new));
+
+ // Put data in batch because it will be logged in form of `DataRecord(List<DataEntry)` on backup node.
+ txCache.putAll(batch);
+
+ // Check `DataRecord(List<DataEntry>)` logged.
+ File archive = U.resolveWorkDirectory(
+ U.defaultWorkDirectory(),
+ grid(1).configuration().getDataStorageConfiguration().getWalArchivePath(),
+ false
+ );
+
+ IteratorParametersBuilder param = new IteratorParametersBuilder().filesOrDirs(archive)
+ .filter((type, pointer) -> type == WALRecord.RecordType.DATA_RECORD_V2);
+
+ assertTrue("DataRecord(List<DataEntry>) should be logged.", waitForCondition(() -> {
+ try (WALIterator iter = new IgniteWalIteratorFactory(log).iterator(param)) {
+ while (iter.hasNext()) {
+ DataRecord rec = (DataRecord)iter.next().get2();
+
+ if (rec.entryCount() > 1)
+ return true;
+ }
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+
+ return false;
+ }, getTestTimeout()));
+
+ for (int i = 0; i < 2; i++) {
+ IgniteEx grid = grid(i);
+
+ Set<Integer> data = new HashSet<>();
+
+ AtomicBoolean firstEvt = new AtomicBoolean(true);
+
+ CdcConsumer cnsmr = new CdcConsumer() {
+ @Override public boolean onEvents(Iterator<CdcEvent> evts) {
+ if (!firstEvt.get())
+ throw new RuntimeException("Expected fail.");
+
+ assertTrue(evts.hasNext());
+
+ data.add((Integer)evts.next().key());
+
+ firstEvt.set(false);
+
+ if (data.size() == keysCnt)
+ throw new RuntimeException("Expected fail.");
+
+ return true;
+ }
+
+ @Override public void stop() {
+ // No-op.
+ }
+
+ @Override public void start(MetricRegistry mreg) {
+ // No-op.
+ }
+ };
+
+ for (int j = 0; j < keysCnt; j++) {
+ CdcMain cdc = createCdc(cnsmr, getConfiguration(grid.name()));
+
+ IgniteInternalFuture<?> fut = runAsync(cdc);
+
+ // Restart CDC after read and commit single key.
+ assertTrue(waitForCondition(fut::isDone, getTestTimeout()));
+ assertEquals(j + 1, data.size());
+
+ firstEvt.set(true);
+ }
+
+ assertTrue(F.eqNotOrdered(batch.keySet(), data));
+ }
+ }
+
+ /** */
+ @Test
public void testReadBeforeGracefulShutdown() throws Exception {
Ignite ign = startGrid(getConfiguration("ignite-0"));