This is an automated email from the ASF dual-hosted git repository.
nizhikov 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 f5bc7fe43ce IGNITE-18652 Fix segments not removed from CDC in case no
DataEntry (#10510)
f5bc7fe43ce is described below
commit f5bc7fe43cea4c28a4b3502746d0f39098033595
Author: Nikolay <[email protected]>
AuthorDate: Wed Feb 1 17:46:10 2023 +0300
IGNITE-18652 Fix segments not removed from CDC in case no DataEntry (#10510)
---
.../org/apache/ignite/internal/cdc/CdcMain.java | 9 +-
.../ignite/internal/cdc/WalRecordsConsumer.java | 4 +-
.../java/org/apache/ignite/cdc/CdcSelfTest.java | 17 +-
.../ignite/internal/cdc/CdcIndexRebuildTest.java | 204 +++++++++++++++++++++
.../IgniteBinaryCacheQueryTestSuite3.java | 5 +-
5 files changed, 225 insertions(+), 14 deletions(-)
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 b7d449fa6a8..35d5a82023c 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
@@ -525,15 +525,16 @@ public class CdcMain implements Runnable {
walState = null;
}
- boolean interrupted = Thread.interrupted();
+ boolean interrupted = false;
- while (iter.hasNext() && !interrupted) {
+ do {
boolean commit = consumer.onRecords(iter);
if (commit) {
T2<WALPointer, Integer> curState = iter.state();
- assert curState != null;
+ if (curState == null)
+ continue;
if (log.isDebugEnabled())
log.debug("Saving state [curState=" + curState + ']');
@@ -560,7 +561,7 @@ public class CdcMain implements Runnable {
}
interrupted = Thread.interrupted();
- }
+ } while (iter.hasNext() && !interrupted);
if (interrupted)
throw new IgniteException("Change Data Capture Application
interrupted");
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 4fd7421afd1..3d450d6cd61 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
@@ -247,7 +247,9 @@ public class WalRecordsConsumer<K, V> {
T2<WALPointer, Integer> state() {
return hasNext() ?
new T2<>(curRec.get1(), entryIdx) :
- new T2<>(curRec.get1().next(), 0);
+ curRec != null
+ ? new T2<>(curRec.get1().next(), 0)
+ : walIter.lastRead().map(ptr -> new T2<>(ptr.next(),
0)).orElse(null);
}
/** Initialize state. */
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 6e748c2eb93..a05d89fccc9 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
@@ -88,9 +88,6 @@ public class CdcSelfTest extends AbstractCdcTest {
/** */
public static final String TX_CACHE_NAME = "tx-cache";
- /** */
- public static final int WAL_ARCHIVE_TIMEOUT = 5_000;
-
/** */
@Parameterized.Parameter
public boolean specificConsistentId;
@@ -157,7 +154,8 @@ public class CdcSelfTest extends AbstractCdcTest {
// Read one record per call.
readAll(new UserCdcConsumer() {
@Override public boolean onEvents(Iterator<CdcEvent> evts) {
- super.onEvents(Collections.singleton(evts.next()).iterator());
+ if (evts.hasNext())
+
super.onEvents(Collections.singleton(evts.next()).iterator());
return false;
}
@@ -170,7 +168,8 @@ public class CdcSelfTest extends AbstractCdcTest {
// Read one record per call and commit.
readAll(new UserCdcConsumer() {
@Override public boolean onEvents(Iterator<CdcEvent> evts) {
- super.onEvents(Collections.singleton(evts.next()).iterator());
+ if (evts.hasNext())
+
super.onEvents(Collections.singleton(evts.next()).iterator());
return true;
}
@@ -275,11 +274,12 @@ public class CdcSelfTest extends AbstractCdcTest {
CdcConsumer cnsmr = new CdcConsumer() {
@Override public boolean onEvents(Iterator<CdcEvent> evts) {
+ if (!evts.hasNext())
+ return true;
+
if (!firstEvt.get())
throw new RuntimeException("Expected fail.");
- assertTrue(evts.hasNext());
-
data.add((Integer)evts.next().key());
firstEvt.set(false);
@@ -363,6 +363,9 @@ public class CdcSelfTest extends AbstractCdcTest {
boolean oneConsumed;
@Override public boolean onEvents(Iterator<CdcEvent> evts) {
+ if (!evts.hasNext())
+ return true;
+
// Fail application after one event read AND state
committed.
if (oneConsumed)
throw new RuntimeException(errMsg);
diff --git
a/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/CdcIndexRebuildTest.java
b/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/CdcIndexRebuildTest.java
new file mode 100644
index 00000000000..eab9766103a
--- /dev/null
+++
b/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/CdcIndexRebuildTest.java
@@ -0,0 +1,204 @@
+/*
+ * 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.cdc;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryType;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.cdc.AbstractCdcTest;
+import org.apache.ignite.cdc.CdcCacheEvent;
+import org.apache.ignite.cdc.CdcConsumer;
+import org.apache.ignite.cdc.CdcEvent;
+import org.apache.ignite.cdc.TypeMapping;
+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.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.processors.metric.MetricRegistry;
+import org.apache.ignite.spi.metric.LongMetric;
+import org.junit.Test;
+
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.internal.cdc.CdcMain.COMMITTED_SEG_IDX;
+import static org.apache.ignite.internal.cdc.CdcMain.CUR_SEG_IDX;
+import static
org.apache.ignite.internal.processors.cache.GridCacheUtils.cacheId;
+import static org.apache.ignite.internal.util.IgniteUtils.MB;
+import static org.apache.ignite.testframework.GridTestUtils.getFieldValue;
+import static org.apache.ignite.testframework.GridTestUtils.runAsync;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** */
+public class CdcIndexRebuildTest extends AbstractCdcTest {
+ /** */
+ public static final int VALS_CNT = 1024 * 30;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ return
super.getConfiguration(igniteInstanceName).setDataStorageConfiguration(new
DataStorageConfiguration()
+ .setWalForceArchiveTimeout(WAL_ARCHIVE_TIMEOUT)
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration()
+ .setPersistenceEnabled(true)
+ .setCdcEnabled(true))
+ .setWalSegmentSize((int)(16 * MB))
+ );
+ }
+
+ /** */
+ @Test
+ public void testIndexRebuild() throws Exception {
+ IgniteEx srv = startGrid(0);
+
+ srv.cluster().state(ACTIVE);
+
+ IgniteCache<Integer, TestVal> cache = srv.getOrCreateCache(
+ new CacheConfiguration<Integer, TestVal>(DEFAULT_CACHE_NAME)
+ .setIndexedTypes(Integer.class, TestVal.class)
+ );
+
+ Map<Integer, TestVal> vals = new HashMap<>(VALS_CNT);
+
+ for (int i = 0; i < VALS_CNT; i++)
+ vals.put(i, new TestVal());
+
+ cache.putAll(vals);
+
+ CountingCdcConsumer cnsmr = new CountingCdcConsumer();
+
+ CdcMain cdc = createCdc(cnsmr, getConfiguration(srv.name()));
+
+ IgniteInternalFuture<Object> fut = runAsync(cdc);
+
+ assertTrue(waitForCondition(() -> cnsmr.commited && cnsmr.evtsCnt ==
VALS_CNT, getTestTimeout()));
+
+ waitForWalSegmentsHandling(srv, cdc);
+
+ cnsmr.reset();
+
+ forceRebuildIndexes(srv, srv.cachex(DEFAULT_CACHE_NAME).context());
+
+ indexRebuildFuture(srv,
cacheId(DEFAULT_CACHE_NAME)).get(getTestTimeout());
+
+ waitForWalSegmentsHandling(srv, cdc);
+
+ fut.cancel();
+ }
+
+ /** @param cdc Cdc. */
+ private void waitForWalSegmentsHandling(IgniteEx srv, CdcMain cdc) throws
IgniteInterruptedCheckedException {
+ assertTrue("Wal segments was not committed by CdcConsumer",
waitForCondition(() -> {
+ long lastArchivedSeg =
srv.context().cache().context().wal().lastArchivedSegment();
+
+ MetricRegistry mreg = getFieldValue(cdc, "mreg");
+
+ long committedCdcSegIdx =
mreg.<LongMetric>findMetric(COMMITTED_SEG_IDX).value();
+ long curCdcSegIdx =
mreg.<LongMetric>findMetric(CUR_SEG_IDX).value();
+
+ log.warning(String.format(">>>>>> Information about CDC and WAL: "
+
+ "[lastArchivedSeg=%d, committedCdcSegIdx=%d,
curCdcSegIdx=%d]",
+ lastArchivedSeg,
+ committedCdcSegIdx,
+ curCdcSegIdx));
+
+ return lastArchivedSeg == committedCdcSegIdx;
+ }, WAL_ARCHIVE_TIMEOUT * 4, WAL_ARCHIVE_TIMEOUT / 2));
+ }
+
+ /** */
+ public static class CountingCdcConsumer implements CdcConsumer {
+ /** Commited. */
+ volatile boolean commited;
+
+ /** Evts count. */
+ volatile int evtsCnt;
+
+ /** */
+ public void reset() {
+ commited = false;
+ evtsCnt = 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start(MetricRegistry mreg) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onEvents(Iterator<CdcEvent> events) {
+ while (events.hasNext()) {
+ events.next();
+
+ evtsCnt++;
+ }
+
+ commited = true;
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onTypes(Iterator<BinaryType> types) {
+ types.forEachRemaining(t -> { /* No-op */ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMappings(Iterator<TypeMapping> mappings) {
+ mappings.forEachRemaining(m -> { /* No-op */ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onCacheChange(Iterator<CdcCacheEvent>
cacheEvents) {
+ cacheEvents.forEachRemaining(e -> { /* No-op */ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onCacheDestroy(Iterator<Integer> caches) {
+ caches.forEachRemaining(c -> { /* No-op */ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() {
+ // No-op.
+ }
+ }
+
+ /** */
+ public static class TestVal {
+ /** Field 0. */
+ @QuerySqlField(index = true, inlineSize = 256)
+ private final String f0;
+
+ /** Field 1. */
+ @QuerySqlField(index = true, inlineSize = 256)
+ private final String f1;
+
+ /**
+ * Default constructor.
+ */
+ public TestVal() {
+ f0 = UUID.randomUUID().toString();
+ f1 = UUID.randomUUID().toString();
+ }
+ }
+}
diff --git
a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
index 93eefe1bf72..17dabe3b376 100644
---
a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
+++
b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
@@ -19,6 +19,7 @@ package org.apache.ignite.testsuites;
import
org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexKeyTypeRegistryTest;
import org.apache.ignite.internal.cdc.CacheEventsCdcTest;
+import org.apache.ignite.internal.cdc.CdcIndexRebuildTest;
import org.apache.ignite.internal.cdc.SqlCdcTest;
import org.apache.ignite.internal.metric.SystemViewSelfTest;
import org.apache.ignite.internal.processors.cache.BigEntryQueryTest;
@@ -376,8 +377,8 @@ import org.junit.runners.Suite;
// CDC tests.
SqlCdcTest.class,
- CacheEventsCdcTest.class
-
+ CacheEventsCdcTest.class,
+ CdcIndexRebuildTest.class
})
public class IgniteBinaryCacheQueryTestSuite3 {
}
