You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2021/07/06 19:53:13 UTC
[GitHub] [ignite] sk0x50 commented on a change in pull request #9206: IGNITE-15012 Adaptation of the historical rebalance to the release of WAL segments
sk0x50 commented on a change in pull request #9206:
URL: https://github.com/apache/ignite/pull/9206#discussion_r664812172
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
##########
@@ -2521,9 +2520,10 @@ private String exchangeTimingsLogMessage(String header, List<String> timings) {
if (localReserved != null) {
boolean success = cctx.database().reserveHistoryForPreloading(localReserved);
- // TODO: how to handle?
- if (!success)
- err = new IgniteCheckedException("Could not reserve history");
+ if (!success) {
Review comment:
In my understanding, the in-memory cluster will always print this warning message (please take a look at `IgniteCacheDatabaseSharedManager#reserveHistoryForPreloading()`). It does not make sense in this case, I think.
##########
File path: modules/core/src/test/java/org/apache/ignite/cache/ReleaseSegmentOnHistoricalRebalance.java
##########
@@ -0,0 +1,408 @@
+/*
+ * 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.cache;
+
+import java.lang.reflect.Method;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+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.failure.StopNodeFailureHandler;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManagerImpl;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistoryResult;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointMarkersStorage;
+import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
+import org.apache.ignite.internal.processors.cache.persistence.wal.aware.SegmentAware;
+import org.apache.ignite.internal.util.lang.IgniteThrowableConsumer;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.testframework.GridTestUtils.getFieldValue;
+import static org.apache.ignite.testframework.GridTestUtils.setFieldValue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Testing the release of WAL segments during historical rebalance.
+ */
+@WithSystemProperty(key = IGNITE_PDS_WAL_REBALANCE_THRESHOLD, value = "0")
+public class ReleaseSegmentOnHistoricalRebalance extends GridCommonAbstractTest {
Review comment:
Please add `Test` postfix to the class name. `ReleaseSegmentOnHistoricalRebalanceTest`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
##########
@@ -1656,8 +1656,14 @@ private String cacheInfo(GridCacheContext cacheCtx) {
int partId = e0.getKey();
- assert cctx.wal().reserved(cpEntry.checkpointMark())
- : "WAL segment for checkpoint " + cpEntry + " has not reserved";
+ if (reservedCheckpointMark != null && !cctx.wal().reserved(reservedCheckpointMark)) {
+ log.warning("Reservation failed because the segment was released: " + reservedCheckpointMark);
+
+ reservedForExchange = null;
+
+ grpPartsWithCnts.clear();
Review comment:
Hmm... Do we need to clear `grpPartsWithCnts` map completely and just switch to the next cache group?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
##########
@@ -1656,8 +1656,14 @@ private String cacheInfo(GridCacheContext cacheCtx) {
int partId = e0.getKey();
- assert cctx.wal().reserved(cpEntry.checkpointMark())
- : "WAL segment for checkpoint " + cpEntry + " has not reserved";
+ if (reservedCheckpointMark != null && !cctx.wal().reserved(reservedCheckpointMark)) {
+ log.warning("Reservation failed because the segment was released: " + reservedCheckpointMark);
Review comment:
Well, the `reservedCheckpointMark` is not changed after initialization at line 1639.
However, if I am not mistaken, it can be "released" at any time by https://issues.apache.org/jira/browse/IGNITE-14952 Perhaps, I am wrong, and we assume that this method `reserveHistoryForExchange` is called from the exchange thread (GridDhtPartitionsExchangeFuture#distributedExchange) and so, WAL segments cannot be released at this point.
Anyway, let's assume that `reservedCheckpointMark` was released right after its initialization. In that case, the warning message will be printed for all cache groups that were applicable for historical rebalance. Maybe I am missing something...
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
##########
@@ -2521,9 +2520,10 @@ private String exchangeTimingsLogMessage(String header, List<String> timings) {
if (localReserved != null) {
boolean success = cctx.database().reserveHistoryForPreloading(localReserved);
- // TODO: how to handle?
- if (!success)
- err = new IgniteCheckedException("Could not reserve history");
+ if (!success) {
+ log.warning("Could not reserve history for historical rebalance " +
+ "(possible it happened because WAL space is exhausted)");
Review comment:
Please add `.` at the end of the sentence :)
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointHistoryResult.java
##########
@@ -58,9 +59,11 @@ public CheckpointHistoryResult(
}
/**
- * @return Reserved checkpoint.
+ * Getting the oldest reserved checkpoint marker.
Review comment:
The common practice is using the Present Simple for writing Javadoc. `Returns the oldest reserved checkpoint marker` looks more natural to me.
##########
File path: modules/core/src/test/java/org/apache/ignite/cache/ReleaseSegmentOnHistoricalRebalance.java
##########
@@ -0,0 +1,408 @@
+/*
+ * 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.cache;
+
+import java.lang.reflect.Method;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+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.failure.StopNodeFailureHandler;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManagerImpl;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistoryResult;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointMarkersStorage;
+import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
+import org.apache.ignite.internal.processors.cache.persistence.wal.aware.SegmentAware;
+import org.apache.ignite.internal.util.lang.IgniteThrowableConsumer;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.testframework.GridTestUtils.getFieldValue;
+import static org.apache.ignite.testframework.GridTestUtils.setFieldValue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Testing the release of WAL segments during historical rebalance.
+ */
+@WithSystemProperty(key = IGNITE_PDS_WAL_REBALANCE_THRESHOLD, value = "0")
+public class ReleaseSegmentOnHistoricalRebalance extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ stopAllGrids();
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setFailureHandler(new StopNodeFailureHandler())
+ .setDataStorageConfiguration(
+ new DataStorageConfiguration()
+ .setWalSegmentSize((int)(2 * U.MB))
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration().setPersistenceEnabled(true))
+ ).setCacheConfiguration(
+ new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ .setAffinity(new RendezvousAffinityFunction(false, 2))
+ .setBackups(1)
+ );
+ }
+
+ /**
+ * Checks that if release the segment after {@link CheckpointHistory#searchAndReserveCheckpoints},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseSegmentAfterSearchAndReserveCheckpoints() throws Exception {
+ checkHistoricalRebalance(n -> {
+ CheckpointHistory spy = spy(dbMgr(n).checkpointHistory());
+
+ when(spy.searchAndReserveCheckpoints(any())).thenAnswer(m -> {
+ CheckpointHistoryResult res = (CheckpointHistoryResult)m.callRealMethod();
+
+ WALPointer reserved = res.reservedCheckpointMark();
+ assertNotNull(reserved);
+
+ release(n, reserved);
+
+ return res;
+ });
+
+ cpHistory(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link GridCacheDatabaseSharedManager#releaseHistoryForExchange},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeReleaseHistoryForExchange() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ doAnswer(m -> {
+ release(n, getFieldValue(spy, "reservedForExchange"));
+
+ return m.callRealMethod();
+ }).when(spy).releaseHistoryForExchange();
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if there is no segment reservation in {@link GridCacheDatabaseSharedManager#reserveHistoryForPreloading},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNoReserveHistoryForPreloading() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ when(spy.reserveHistoryForPreloading(any())).thenAnswer(m -> false);
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link GridCacheDatabaseSharedManager#releaseHistoryForPreloading},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeReleaseHistoryForPreloading() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ doAnswer(m -> {
+ release(n, spy.latestWalPointerReservedForPreloading());
+
+ return m.callRealMethod();
+ }).when(spy).releaseHistoryForPreloading();
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link IgniteCacheOffheapManagerImpl#rebalanceIterator},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeRebalanceIterator() throws Exception {
+ checkHistoricalRebalance(n -> {
+ IgniteInternalCache cachex = n.cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheOffheapManager spy = spy(offheapMgr(cachex));
+
+ doAnswer(m -> {
+ CheckpointHistory cpHist = dbMgr(n).checkpointHistory();
+
+ for (Long cp : cpHist.checkpoints())
+ release(n, entry(cpHist, cp).checkpointMark());
+
+ return m.callRealMethod();
+ }).when(spy).rebalanceIterator(any(), any());
+
+ offheapMgr(cachex, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment during {@link IgniteCacheOffheapManagerImpl#rebalanceIterator},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseDuringRebalanceIterator() throws Exception {
+ checkHistoricalRebalance(n -> {
+ IgniteInternalCache cachex = n.cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheOffheapManager spy = spy(offheapMgr(cachex));
+
+ doAnswer(m -> {
+ WALPointer reserved = dbMgr(n).latestWalPointerReservedForPreloading();
+
+ assertNotNull(reserved);
+
+ Object o = m.callRealMethod();
+
+ release(n, reserved);
+
+ assertTrue(segmentAware(n).minReserveIndex(Long.MAX_VALUE));
+
+ return o;
+ }).when(spy).rebalanceIterator(any(), any());
+
+ offheapMgr(cachex, spy);
+ });
+ }
+
+ /**
+ * Checks that if the reservation is released immediately,
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testImmediateReleaseSegment() throws Exception {
+ checkHistoricalRebalance(n -> {
+ SegmentAware spy = spy(segmentAware(n));
+
+ doAnswer(m -> {
+ Object o = m.callRealMethod();
+
+ spy.release(m.getArgument(0));
+
+ return o;
+ }).when(spy).reserve(anyLong());
+
+ segmentAware(n, spy);
+ });
+ }
+
+ /**
+ * Checks that that if there is no reservation,
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNoReserveSegment() throws Exception {
+ checkHistoricalRebalance(n -> {
+ SegmentAware spy = spy(segmentAware(n));
+
+ when(spy.reserve(anyLong())).thenAnswer(m -> false);
+
+ segmentAware(n, spy);
+ });
+ }
+
+ /**
+ * Populate the cache.
+ *
+ * @param cache Cache.
+ * @param cnt Entry count.
+ * @param o Key offset.
+ * @throws Exception If failed.
+ */
+ private void populate(IgniteCache<? super Object, ? super Object> cache, int cnt, int o) throws Exception {
+ for (int i = 0; i < cnt; i++) {
+ if (i % 100 == 0)
+ forceCheckpoint();
+
+ cache.put(i + o, new byte[64 * 1024]);
+ }
+ }
+
+ /**
+ * Set the spy to {@code CheckpointMarkersStorage#cpHistory}.
+ *
+ * @param n Node.
+ * @param spy Spy.
+ */
+ private void cpHistory(IgniteEx n, CheckpointHistory spy) {
+ CheckpointMarkersStorage s = getFieldValue(dbMgr(n), "checkpointManager", "checkpointMarkersStorage");
+
+ setFieldValue(s, "cpHistory", spy);
+ }
+
+ /**
+ * Set the spy to {@code GridCacheSharedContext#dbMgr}.
+ *
+ * @param n Node.
+ * @param spy Spy.
+ */
+ private void dbMgr(IgniteEx n, GridCacheDatabaseSharedManager spy) {
Review comment:
Please do not abbreviate method names https://cwiki.apache.org/confluence/display/IGNITE/Coding+Guidelines#CodingGuidelines-Abbreviations
##########
File path: modules/core/src/test/java/org/apache/ignite/cache/ReleaseSegmentOnHistoricalRebalance.java
##########
@@ -0,0 +1,408 @@
+/*
+ * 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.cache;
+
+import java.lang.reflect.Method;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+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.failure.StopNodeFailureHandler;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManagerImpl;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistoryResult;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointMarkersStorage;
+import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
+import org.apache.ignite.internal.processors.cache.persistence.wal.aware.SegmentAware;
+import org.apache.ignite.internal.util.lang.IgniteThrowableConsumer;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.testframework.GridTestUtils.getFieldValue;
+import static org.apache.ignite.testframework.GridTestUtils.setFieldValue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Testing the release of WAL segments during historical rebalance.
+ */
+@WithSystemProperty(key = IGNITE_PDS_WAL_REBALANCE_THRESHOLD, value = "0")
+public class ReleaseSegmentOnHistoricalRebalance extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ stopAllGrids();
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setFailureHandler(new StopNodeFailureHandler())
+ .setDataStorageConfiguration(
+ new DataStorageConfiguration()
+ .setWalSegmentSize((int)(2 * U.MB))
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration().setPersistenceEnabled(true))
+ ).setCacheConfiguration(
+ new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ .setAffinity(new RendezvousAffinityFunction(false, 2))
+ .setBackups(1)
+ );
+ }
+
+ /**
+ * Checks that if release the segment after {@link CheckpointHistory#searchAndReserveCheckpoints},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseSegmentAfterSearchAndReserveCheckpoints() throws Exception {
+ checkHistoricalRebalance(n -> {
+ CheckpointHistory spy = spy(dbMgr(n).checkpointHistory());
+
+ when(spy.searchAndReserveCheckpoints(any())).thenAnswer(m -> {
+ CheckpointHistoryResult res = (CheckpointHistoryResult)m.callRealMethod();
+
+ WALPointer reserved = res.reservedCheckpointMark();
+ assertNotNull(reserved);
+
+ release(n, reserved);
+
+ return res;
+ });
+
+ cpHistory(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link GridCacheDatabaseSharedManager#releaseHistoryForExchange},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeReleaseHistoryForExchange() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ doAnswer(m -> {
+ release(n, getFieldValue(spy, "reservedForExchange"));
+
+ return m.callRealMethod();
+ }).when(spy).releaseHistoryForExchange();
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if there is no segment reservation in {@link GridCacheDatabaseSharedManager#reserveHistoryForPreloading},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNoReserveHistoryForPreloading() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ when(spy.reserveHistoryForPreloading(any())).thenAnswer(m -> false);
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link GridCacheDatabaseSharedManager#releaseHistoryForPreloading},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeReleaseHistoryForPreloading() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ doAnswer(m -> {
+ release(n, spy.latestWalPointerReservedForPreloading());
+
+ return m.callRealMethod();
+ }).when(spy).releaseHistoryForPreloading();
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link IgniteCacheOffheapManagerImpl#rebalanceIterator},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeRebalanceIterator() throws Exception {
+ checkHistoricalRebalance(n -> {
+ IgniteInternalCache cachex = n.cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheOffheapManager spy = spy(offheapMgr(cachex));
+
+ doAnswer(m -> {
+ CheckpointHistory cpHist = dbMgr(n).checkpointHistory();
+
+ for (Long cp : cpHist.checkpoints())
+ release(n, entry(cpHist, cp).checkpointMark());
+
+ return m.callRealMethod();
+ }).when(spy).rebalanceIterator(any(), any());
+
+ offheapMgr(cachex, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment during {@link IgniteCacheOffheapManagerImpl#rebalanceIterator},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseDuringRebalanceIterator() throws Exception {
+ checkHistoricalRebalance(n -> {
+ IgniteInternalCache cachex = n.cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheOffheapManager spy = spy(offheapMgr(cachex));
+
+ doAnswer(m -> {
+ WALPointer reserved = dbMgr(n).latestWalPointerReservedForPreloading();
+
+ assertNotNull(reserved);
+
+ Object o = m.callRealMethod();
+
+ release(n, reserved);
+
+ assertTrue(segmentAware(n).minReserveIndex(Long.MAX_VALUE));
+
+ return o;
+ }).when(spy).rebalanceIterator(any(), any());
+
+ offheapMgr(cachex, spy);
+ });
+ }
+
+ /**
+ * Checks that if the reservation is released immediately,
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testImmediateReleaseSegment() throws Exception {
+ checkHistoricalRebalance(n -> {
+ SegmentAware spy = spy(segmentAware(n));
+
+ doAnswer(m -> {
+ Object o = m.callRealMethod();
+
+ spy.release(m.getArgument(0));
+
+ return o;
+ }).when(spy).reserve(anyLong());
+
+ segmentAware(n, spy);
+ });
+ }
+
+ /**
+ * Checks that that if there is no reservation,
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNoReserveSegment() throws Exception {
+ checkHistoricalRebalance(n -> {
+ SegmentAware spy = spy(segmentAware(n));
+
+ when(spy.reserve(anyLong())).thenAnswer(m -> false);
+
+ segmentAware(n, spy);
+ });
+ }
+
+ /**
+ * Populate the cache.
Review comment:
`Populates the given cache and forces a new checkpoint every 100 updates.` What do you think?
##########
File path: modules/core/src/test/java/org/apache/ignite/cache/ReleaseSegmentOnHistoricalRebalance.java
##########
@@ -0,0 +1,408 @@
+/*
+ * 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.cache;
+
+import java.lang.reflect.Method;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+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.failure.StopNodeFailureHandler;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManagerImpl;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistoryResult;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointMarkersStorage;
+import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
+import org.apache.ignite.internal.processors.cache.persistence.wal.aware.SegmentAware;
+import org.apache.ignite.internal.util.lang.IgniteThrowableConsumer;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_PDS_WAL_REBALANCE_THRESHOLD;
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.testframework.GridTestUtils.getFieldValue;
+import static org.apache.ignite.testframework.GridTestUtils.setFieldValue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Testing the release of WAL segments during historical rebalance.
+ */
+@WithSystemProperty(key = IGNITE_PDS_WAL_REBALANCE_THRESHOLD, value = "0")
+public class ReleaseSegmentOnHistoricalRebalance extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ stopAllGrids();
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setFailureHandler(new StopNodeFailureHandler())
+ .setDataStorageConfiguration(
+ new DataStorageConfiguration()
+ .setWalSegmentSize((int)(2 * U.MB))
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration().setPersistenceEnabled(true))
+ ).setCacheConfiguration(
+ new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ .setAffinity(new RendezvousAffinityFunction(false, 2))
+ .setBackups(1)
+ );
+ }
+
+ /**
+ * Checks that if release the segment after {@link CheckpointHistory#searchAndReserveCheckpoints},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseSegmentAfterSearchAndReserveCheckpoints() throws Exception {
+ checkHistoricalRebalance(n -> {
+ CheckpointHistory spy = spy(dbMgr(n).checkpointHistory());
+
+ when(spy.searchAndReserveCheckpoints(any())).thenAnswer(m -> {
+ CheckpointHistoryResult res = (CheckpointHistoryResult)m.callRealMethod();
+
+ WALPointer reserved = res.reservedCheckpointMark();
+ assertNotNull(reserved);
+
+ release(n, reserved);
+
+ return res;
+ });
+
+ cpHistory(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link GridCacheDatabaseSharedManager#releaseHistoryForExchange},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeReleaseHistoryForExchange() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ doAnswer(m -> {
+ release(n, getFieldValue(spy, "reservedForExchange"));
+
+ return m.callRealMethod();
+ }).when(spy).releaseHistoryForExchange();
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if there is no segment reservation in {@link GridCacheDatabaseSharedManager#reserveHistoryForPreloading},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNoReserveHistoryForPreloading() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ when(spy.reserveHistoryForPreloading(any())).thenAnswer(m -> false);
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link GridCacheDatabaseSharedManager#releaseHistoryForPreloading},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeReleaseHistoryForPreloading() throws Exception {
+ checkHistoricalRebalance(n -> {
+ GridCacheDatabaseSharedManager spy = spy(dbMgr(n));
+
+ doAnswer(m -> {
+ release(n, spy.latestWalPointerReservedForPreloading());
+
+ return m.callRealMethod();
+ }).when(spy).releaseHistoryForPreloading();
+
+ dbMgr(n, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment before {@link IgniteCacheOffheapManagerImpl#rebalanceIterator},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseBeforeRebalanceIterator() throws Exception {
+ checkHistoricalRebalance(n -> {
+ IgniteInternalCache cachex = n.cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheOffheapManager spy = spy(offheapMgr(cachex));
+
+ doAnswer(m -> {
+ CheckpointHistory cpHist = dbMgr(n).checkpointHistory();
+
+ for (Long cp : cpHist.checkpoints())
+ release(n, entry(cpHist, cp).checkpointMark());
+
+ return m.callRealMethod();
+ }).when(spy).rebalanceIterator(any(), any());
+
+ offheapMgr(cachex, spy);
+ });
+ }
+
+ /**
+ * Checks that if release the segment during {@link IgniteCacheOffheapManagerImpl#rebalanceIterator},
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testReleaseDuringRebalanceIterator() throws Exception {
+ checkHistoricalRebalance(n -> {
+ IgniteInternalCache cachex = n.cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheOffheapManager spy = spy(offheapMgr(cachex));
+
+ doAnswer(m -> {
+ WALPointer reserved = dbMgr(n).latestWalPointerReservedForPreloading();
+
+ assertNotNull(reserved);
+
+ Object o = m.callRealMethod();
+
+ release(n, reserved);
+
+ assertTrue(segmentAware(n).minReserveIndex(Long.MAX_VALUE));
+
+ return o;
+ }).when(spy).rebalanceIterator(any(), any());
+
+ offheapMgr(cachex, spy);
+ });
+ }
+
+ /**
+ * Checks that if the reservation is released immediately,
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testImmediateReleaseSegment() throws Exception {
+ checkHistoricalRebalance(n -> {
+ SegmentAware spy = spy(segmentAware(n));
+
+ doAnswer(m -> {
+ Object o = m.callRealMethod();
+
+ spy.release(m.getArgument(0));
+
+ return o;
+ }).when(spy).reserve(anyLong());
+
+ segmentAware(n, spy);
+ });
+ }
+
+ /**
+ * Checks that that if there is no reservation,
+ * there will be no errors and the rebalance will be completed.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNoReserveSegment() throws Exception {
+ checkHistoricalRebalance(n -> {
+ SegmentAware spy = spy(segmentAware(n));
+
+ when(spy.reserve(anyLong())).thenAnswer(m -> false);
+
+ segmentAware(n, spy);
+ });
+ }
+
+ /**
+ * Populate the cache.
+ *
+ * @param cache Cache.
+ * @param cnt Entry count.
+ * @param o Key offset.
+ * @throws Exception If failed.
+ */
+ private void populate(IgniteCache<? super Object, ? super Object> cache, int cnt, int o) throws Exception {
Review comment:
well, the key offset for the type `? super Object` looks odd to me :) Perhaps, it can be changed to `IgniteCache<Integer, ? super Object> cache`. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org