modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsEntitiesCountTest.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;

 import java.util.ArrayList;
 import java.util.Collection;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.cache.CacheAtomicityMode;
 import org.apache.ignite.cache.CacheMetrics;
 import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cache.CachePeekMode;
 import org.apache.ignite.cache.CacheWriteSynchronizationMode;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.configuration.NearCacheConfiguration;
 import org.apache.ignite.internal.processors.metric.MetricRegistry;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.spi.metric.LongMetric;
 import org.apache.ignite.testframework.MvccFeatureChecker;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;

 import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.cacheMetricsRegistryName;

 /**
  * This test checks that entries count metrics, calculated by method
  * {@link org.apache.ignite.internal.processors.cache.CacheMetricsImpl#getEntriesStat()} (which uses one iteration
  * over local partitions to get all set of metrics), have the same values as metrics, calculated by individual methods
  * (which use iteration over local partition per each method call).
  */
 public class CacheMetricsEntitiesCountTest extends GridCommonAbstractTest {
     /** Grid count. */
     private static final int GRID_CNT = 3;

     /** Cache prefix. */
     private static final String CACHE_PREFIX = "CACHE";

     /** Entities cnt. */
     private static final int ENTITIES_CNT = 100;

     /** Onheap peek modes. */
     private static final CachePeekMode[] ONHEAP_PEEK_MODES = new CachePeekMode[] {
         CachePeekMode.ONHEAP, CachePeekMode.PRIMARY, CachePeekMode.BACKUP, CachePeekMode.NEAR};

     /** Cache count. */
     private static int cacheCnt = 4;

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

         Collection<CacheConfiguration> ccfgs = new ArrayList<>(4);

         ccfgs.add(new CacheConfiguration<>()
             .setName(CACHE_PREFIX + 0)
             .setStatisticsEnabled(true)
             .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
             .setCacheMode(CacheMode.REPLICATED)
             .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

         ccfgs.add(new CacheConfiguration<>()
             .setName(CACHE_PREFIX + 1)
             .setStatisticsEnabled(true)
             .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
             .setCacheMode(CacheMode.PARTITIONED)
             .setBackups(1)
             .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

         ccfgs.add(new CacheConfiguration<>()
             .setName(CACHE_PREFIX + 2)
             .setStatisticsEnabled(true)
             .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
             .setCacheMode(CacheMode.PARTITIONED)
             .setBackups(1)
             .setNearConfiguration(new NearCacheConfiguration<>())
             .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

         ccfgs.add(new CacheConfiguration<>()
             .setName(CACHE_PREFIX + 3)
             .setStatisticsEnabled(true)
             .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
             .setCacheMode(CacheMode.PARTITIONED)
             .setBackups(1)
             .setOnheapCacheEnabled(true)
             .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

         if (!MvccFeatureChecker.forcedMvcc() || MvccFeatureChecker.isSupported(MvccFeatureChecker.Feature.LOCAL_CACHE)) {
             ccfgs.add(new CacheConfiguration<>()
                 .setName(CACHE_PREFIX + 4)
                 .setStatisticsEnabled(true)
                 .setCacheMode(CacheMode.LOCAL)
                 .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));
         }

         cacheCnt = ccfgs.size();

         cfg.setCacheConfiguration(U.toArray(ccfgs, new CacheConfiguration[cacheCnt]));

         return cfg;
     }

     /** {@inheritDoc} */
     @Override protected void beforeTestsStarted() throws Exception {
         startGrids(GRID_CNT);
     }

     /**
      * Test entities count, calculated by different implementations.
      */
     @Test
     public void testEnitiesCount() throws Exception {
         awaitPartitionMapExchange();

         for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++)
             for (int cacheIdx = 0; cacheIdx < cacheCnt; cacheIdx++)
                 fillCache(igniteIdx, cacheIdx);

         awaitMetricsUpdate(1);

         int cacheSize = GRID_CNT * ENTITIES_CNT;

         // CacheMode == REPLICATED.
         checkCacheClusterMetrics(0,
             cacheSize,
             cacheSize * GRID_CNT,
             cacheSize,
             cacheSize * (GRID_CNT - 1),
             0);

         // CacheMode == PARTITIONED, Backups == 1.
         checkCacheClusterMetrics(1,
             cacheSize,
             cacheSize * 2,
             cacheSize,
             cacheSize,
             0);

         // CacheMode == PARTITIONED, Backups == 1, NearCache.
         checkCacheClusterMetrics(2,
             cacheSize,
             cacheSize * 2,
             cacheSize,
             cacheSize,
             216 /* TODO */);

         // CacheMode == PARTITIONED, Backups == 1, OnheapCache.
         checkCacheClusterMetrics(3,
             cacheSize,
             cacheSize * 2,
             cacheSize,
             cacheSize,
             cacheSize * 2);

         // CacheMode == LOCAL
         if (cacheCnt == 5)
             checkCacheClusterMetrics(4,
                 cacheSize,
                 cacheSize,
                 cacheSize,
                 0,
                 0);

         for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++)
             for (int cacheIdx = 0; cacheIdx < cacheCnt; cacheIdx++)
                 checkCacheLocalMetrics(igniteIdx, cacheIdx);
     }

     /**
      * @param igniteIdx Ignite index.
      * @param cacheIdx Cache index.
      */
     private void fillCache(int igniteIdx, int cacheIdx) {
         log.info("Filling cache, igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx);

         IgniteCache cache = grid(igniteIdx).cache(CACHE_PREFIX + cacheIdx);

         for (int i = 0; i < ENTITIES_CNT; i++)
             cache.put("key" + igniteIdx + "-" + i, i);
     }

     /**
      * @param igniteIdx Ignite index.
      * @param cacheIdx Cache index.
      */
     private void checkCacheLocalMetrics(int igniteIdx, int cacheIdx) throws IgniteCheckedException {
         IgniteInternalCache internalCache = grid(igniteIdx).cachex(CACHE_PREFIX + cacheIdx);

         GridCacheContext cctx = internalCache.context();

         GridCacheAdapter cache = cctx.cache();

         CacheMetricsImpl metrics = cache.metrics0();

         long offHeapEntriesCount = cache.offHeapEntriesCount();

         long offHeapPrimaryEntriesCount = cctx.offheap().cacheEntriesCount(cctx.cacheId(),
             true,
             false,
             cctx.affinity().affinityTopologyVersion());

         long offHeapBackupEntriesCount = cctx.offheap().cacheEntriesCount(cctx.cacheId(),
             false,
             true,
             cctx.affinity().affinityTopologyVersion());

         long heapEntriesCount = cache.localSizeLong(ONHEAP_PEEK_MODES);

         long cacheSize = cache.localSizeLong(new CachePeekMode[]{CachePeekMode.PRIMARY});
         int size = cache.localSize(new CachePeekMode[]{CachePeekMode.PRIMARY});

         boolean isEmpty = cache.isEmpty();

         String cacheInfo = "igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx + " ";

         log.info("Checking cache,  " + cacheInfo);

         assertEquals(cacheInfo + " offHeapEntriesCount",
             offHeapEntriesCount, metrics.getOffHeapEntriesCount());
         assertEquals(cacheInfo + " offHeapBackupEntriesCount",
             offHeapBackupEntriesCount, metrics.getOffHeapBackupEntriesCount());
         assertEquals(cacheInfo + " offHeapPrimaryEntriesCount",
             offHeapPrimaryEntriesCount, metrics.getOffHeapPrimaryEntriesCount());
         assertEquals(cacheInfo + " heapEntriesCount", heapEntriesCount, metrics.getHeapEntriesCount());
         assertEquals(cacheInfo + " size", size, metrics.getSize());
         assertEquals(cacheInfo + " keySize", size, metrics.getKeySize());
         assertEquals(cacheInfo + " cacheSize", cacheSize, metrics.getCacheSize());
         assertEquals(cacheInfo + " isEmpty", isEmpty, metrics.isEmpty());

         MetricRegistry mreg = cctx.kernalContext().metric().registry(cacheMetricsRegistryName(cctx.name(),
             cache.isNear()));

         assertNotNull(mreg);

         assertEquals(offHeapEntriesCount, ((LongMetric)mreg.findMetric("OffHeapEntriesCount")).value());
         assertEquals(offHeapBackupEntriesCount, ((LongMetric)mreg.findMetric("OffHeapBackupEntriesCount")).value());
         assertEquals(offHeapPrimaryEntriesCount, ((LongMetric)mreg.findMetric("OffHeapPrimaryEntriesCount")).value());
         assertEquals(heapEntriesCount, ((LongMetric)mreg.findMetric("HeapEntriesCount")).value());
         assertEquals(cacheSize, ((LongMetric)mreg.findMetric("CacheSize")).value());
     }

     /**
      * @param cacheIdx Cache index.
      */
     private void checkCacheClusterMetrics(int cacheIdx,
         long cacheSize,
         long offHeapEntriesCnt,
         long offHeapPrimaryEntriesCnt,
         long offHeapBackupEntriesCnt,
         long heapEntriesCnt
     ) {
         long cacheSizeSum = 0;
         long offHeapEntriesCntSum = 0;
         long offHeapPrimaryEntriesCntSum = 0;
         long offHeapBackupEntriesCntSum = 0;
         long heapEntriesCntSum = 0;
         boolean isEmptySum = true;

         for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++) {
             IgniteCache cache = grid(igniteIdx).cache(CACHE_PREFIX + cacheIdx);

             CacheMetrics metrics = cache.metrics();

             String cacheInfo = "igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx + " ";

             assertEquals(cacheInfo + " CacheSize", cacheSize, metrics.getCacheSize());
             assertEquals(cacheInfo + " offHeapEntriesCnt", offHeapEntriesCnt,
                 metrics.getOffHeapEntriesCount());
             assertEquals(cacheInfo + " offHeapBackupEntriesCnt", offHeapBackupEntriesCnt,
                 metrics.getOffHeapBackupEntriesCount());
             assertEquals(cacheInfo + " offHeapPrimaryEntriesCnt", offHeapPrimaryEntriesCnt,
                 metrics.getOffHeapPrimaryEntriesCount());

             if (!MvccFeatureChecker.forcedMvcc()) // Onheap cache is not supported in Mvcc mode.
                 assertEquals(cacheInfo + " heapEntriesCnt", heapEntriesCnt, metrics.getHeapEntriesCount());

             assertEquals(cacheInfo + " size", cacheSize, metrics.getSize());
             assertEquals(cacheInfo + " keySize", cacheSize, metrics.getKeySize());
             assertEquals(cacheInfo + " isEmpty", cacheSize == 0, metrics.isEmpty());

             metrics = cache.localMetrics();

             cacheSizeSum += metrics.getCacheSize();
             offHeapEntriesCntSum += metrics.getOffHeapEntriesCount();
             offHeapPrimaryEntriesCntSum += metrics.getOffHeapPrimaryEntriesCount();
             offHeapBackupEntriesCntSum += metrics.getOffHeapBackupEntriesCount();
             heapEntriesCntSum += metrics.getHeapEntriesCount();
             isEmptySum = isEmptySum && metrics.isEmpty();
         }

         String cacheInfo = "cacheIdx=" + cacheIdx + " check sum";

         assertEquals(cacheInfo + " CacheSize", cacheSize, cacheSizeSum);
         assertEquals(cacheInfo + " offHeapEntriesCnt", offHeapEntriesCnt, offHeapEntriesCntSum);
         assertEquals(cacheInfo + " offHeapBackupEntriesCnt", offHeapBackupEntriesCnt, offHeapBackupEntriesCntSum);
         assertEquals(cacheInfo + " offHeapPrimaryEntriesCnt", offHeapPrimaryEntriesCnt, offHeapPrimaryEntriesCntSum);

         if (!MvccFeatureChecker.forcedMvcc()) // Onheap cache is not supported in Mvcc mode.
             assertEquals(cacheInfo + " heapEntriesCnt", heapEntriesCnt, heapEntriesCntSum);

         assertEquals(cacheInfo + " isEmpty", cacheSize == 0, isEmptySum);
     }
 }
	/*
	* 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;

	import java.util.ArrayList;
	import java.util.Collection;
	import org.apache.ignite.IgniteCache;
	import org.apache.ignite.IgniteCheckedException;
	import org.apache.ignite.cache.CacheAtomicityMode;
	import org.apache.ignite.cache.CacheMetrics;
	import org.apache.ignite.cache.CacheMode;
	import org.apache.ignite.cache.CachePeekMode;
	import org.apache.ignite.cache.CacheWriteSynchronizationMode;
	import org.apache.ignite.configuration.CacheConfiguration;
	import org.apache.ignite.configuration.IgniteConfiguration;
	import org.apache.ignite.configuration.NearCacheConfiguration;
	import org.apache.ignite.internal.processors.metric.MetricRegistry;
	import org.apache.ignite.internal.util.typedef.internal.U;
	import org.apache.ignite.spi.metric.LongMetric;
	import org.apache.ignite.testframework.MvccFeatureChecker;
	import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
	import org.junit.Test;

	import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.cacheMetricsRegistryName;

	/**
	* This test checks that entries count metrics, calculated by method
	* {@link org.apache.ignite.internal.processors.cache.CacheMetricsImpl#getEntriesStat()} (which uses one iteration
	* over local partitions to get all set of metrics), have the same values as metrics, calculated by individual methods
	* (which use iteration over local partition per each method call).
	*/
	public class CacheMetricsEntitiesCountTest extends GridCommonAbstractTest {
	/** Grid count. */
	private static final int GRID_CNT = 3;

	/** Cache prefix. */
	private static final String CACHE_PREFIX = "CACHE";

	/** Entities cnt. */
	private static final int ENTITIES_CNT = 100;

	/** Onheap peek modes. */
	private static final CachePeekMode[] ONHEAP_PEEK_MODES = new CachePeekMode[] {
	CachePeekMode.ONHEAP, CachePeekMode.PRIMARY, CachePeekMode.BACKUP, CachePeekMode.NEAR};

	/** Cache count. */
	private static int cacheCnt = 4;

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

	Collection<CacheConfiguration> ccfgs = new ArrayList<>(4);

	ccfgs.add(new CacheConfiguration<>()
	.setName(CACHE_PREFIX + 0)
	.setStatisticsEnabled(true)
	.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
	.setCacheMode(CacheMode.REPLICATED)
	.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

	ccfgs.add(new CacheConfiguration<>()
	.setName(CACHE_PREFIX + 1)
	.setStatisticsEnabled(true)
	.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
	.setCacheMode(CacheMode.PARTITIONED)
	.setBackups(1)
	.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

	ccfgs.add(new CacheConfiguration<>()
	.setName(CACHE_PREFIX + 2)
	.setStatisticsEnabled(true)
	.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
	.setCacheMode(CacheMode.PARTITIONED)
	.setBackups(1)
	.setNearConfiguration(new NearCacheConfiguration<>())
	.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

	ccfgs.add(new CacheConfiguration<>()
	.setName(CACHE_PREFIX + 3)
	.setStatisticsEnabled(true)
	.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
	.setCacheMode(CacheMode.PARTITIONED)
	.setBackups(1)
	.setOnheapCacheEnabled(true)
	.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

	if (!MvccFeatureChecker.forcedMvcc() \|\| MvccFeatureChecker.isSupported(MvccFeatureChecker.Feature.LOCAL_CACHE)) {
	ccfgs.add(new CacheConfiguration<>()
	.setName(CACHE_PREFIX + 4)
	.setStatisticsEnabled(true)
	.setCacheMode(CacheMode.LOCAL)
	.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));
	}

	cacheCnt = ccfgs.size();

	cfg.setCacheConfiguration(U.toArray(ccfgs, new CacheConfiguration[cacheCnt]));

	return cfg;
	}

	/** {@inheritDoc} */
	@Override protected void beforeTestsStarted() throws Exception {
	startGrids(GRID_CNT);
	}

	/**
	* Test entities count, calculated by different implementations.
	*/
	@Test
	public void testEnitiesCount() throws Exception {
	awaitPartitionMapExchange();

	for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++)
	for (int cacheIdx = 0; cacheIdx < cacheCnt; cacheIdx++)
	fillCache(igniteIdx, cacheIdx);

	awaitMetricsUpdate(1);

	int cacheSize = GRID_CNT * ENTITIES_CNT;

	// CacheMode == REPLICATED.
	checkCacheClusterMetrics(0,
	cacheSize,
	cacheSize * GRID_CNT,
	cacheSize,
	cacheSize * (GRID_CNT - 1),
	0);

	// CacheMode == PARTITIONED, Backups == 1.
	checkCacheClusterMetrics(1,
	cacheSize,
	cacheSize * 2,
	cacheSize,
	cacheSize,
	0);

	// CacheMode == PARTITIONED, Backups == 1, NearCache.
	checkCacheClusterMetrics(2,
	cacheSize,
	cacheSize * 2,
	cacheSize,
	cacheSize,
	216 /* TODO */);

	// CacheMode == PARTITIONED, Backups == 1, OnheapCache.
	checkCacheClusterMetrics(3,
	cacheSize,
	cacheSize * 2,
	cacheSize,
	cacheSize,
	cacheSize * 2);

	// CacheMode == LOCAL
	if (cacheCnt == 5)
	checkCacheClusterMetrics(4,
	cacheSize,
	cacheSize,
	cacheSize,
	0,
	0);

	for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++)
	for (int cacheIdx = 0; cacheIdx < cacheCnt; cacheIdx++)
	checkCacheLocalMetrics(igniteIdx, cacheIdx);
	}

	/**
	* @param igniteIdx Ignite index.
	* @param cacheIdx Cache index.
	*/
	private void fillCache(int igniteIdx, int cacheIdx) {
	log.info("Filling cache, igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx);

	IgniteCache cache = grid(igniteIdx).cache(CACHE_PREFIX + cacheIdx);

	for (int i = 0; i < ENTITIES_CNT; i++)
	cache.put("key" + igniteIdx + "-" + i, i);
	}

	/**
	* @param igniteIdx Ignite index.
	* @param cacheIdx Cache index.
	*/
	private void checkCacheLocalMetrics(int igniteIdx, int cacheIdx) throws IgniteCheckedException {
	IgniteInternalCache internalCache = grid(igniteIdx).cachex(CACHE_PREFIX + cacheIdx);

	GridCacheContext cctx = internalCache.context();

	GridCacheAdapter cache = cctx.cache();

	CacheMetricsImpl metrics = cache.metrics0();

	long offHeapEntriesCount = cache.offHeapEntriesCount();

	long offHeapPrimaryEntriesCount = cctx.offheap().cacheEntriesCount(cctx.cacheId(),
	true,
	false,
	cctx.affinity().affinityTopologyVersion());

	long offHeapBackupEntriesCount = cctx.offheap().cacheEntriesCount(cctx.cacheId(),
	false,
	true,
	cctx.affinity().affinityTopologyVersion());

	long heapEntriesCount = cache.localSizeLong(ONHEAP_PEEK_MODES);

	long cacheSize = cache.localSizeLong(new CachePeekMode[]{CachePeekMode.PRIMARY});
	int size = cache.localSize(new CachePeekMode[]{CachePeekMode.PRIMARY});

	boolean isEmpty = cache.isEmpty();

	String cacheInfo = "igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx + " ";

	log.info("Checking cache, " + cacheInfo);

	assertEquals(cacheInfo + " offHeapEntriesCount",
	offHeapEntriesCount, metrics.getOffHeapEntriesCount());
	assertEquals(cacheInfo + " offHeapBackupEntriesCount",
	offHeapBackupEntriesCount, metrics.getOffHeapBackupEntriesCount());
	assertEquals(cacheInfo + " offHeapPrimaryEntriesCount",
	offHeapPrimaryEntriesCount, metrics.getOffHeapPrimaryEntriesCount());
	assertEquals(cacheInfo + " heapEntriesCount", heapEntriesCount, metrics.getHeapEntriesCount());
	assertEquals(cacheInfo + " size", size, metrics.getSize());
	assertEquals(cacheInfo + " keySize", size, metrics.getKeySize());
	assertEquals(cacheInfo + " cacheSize", cacheSize, metrics.getCacheSize());
	assertEquals(cacheInfo + " isEmpty", isEmpty, metrics.isEmpty());

	MetricRegistry mreg = cctx.kernalContext().metric().registry(cacheMetricsRegistryName(cctx.name(),
	cache.isNear()));

	assertNotNull(mreg);

	assertEquals(offHeapEntriesCount, ((LongMetric)mreg.findMetric("OffHeapEntriesCount")).value());
	assertEquals(offHeapBackupEntriesCount, ((LongMetric)mreg.findMetric("OffHeapBackupEntriesCount")).value());
	assertEquals(offHeapPrimaryEntriesCount, ((LongMetric)mreg.findMetric("OffHeapPrimaryEntriesCount")).value());
	assertEquals(heapEntriesCount, ((LongMetric)mreg.findMetric("HeapEntriesCount")).value());
	assertEquals(cacheSize, ((LongMetric)mreg.findMetric("CacheSize")).value());
	}

	/**
	* @param cacheIdx Cache index.
	*/
	private void checkCacheClusterMetrics(int cacheIdx,
	long cacheSize,
	long offHeapEntriesCnt,
	long offHeapPrimaryEntriesCnt,
	long offHeapBackupEntriesCnt,
	long heapEntriesCnt
	) {
	long cacheSizeSum = 0;
	long offHeapEntriesCntSum = 0;
	long offHeapPrimaryEntriesCntSum = 0;
	long offHeapBackupEntriesCntSum = 0;
	long heapEntriesCntSum = 0;
	boolean isEmptySum = true;

	for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++) {
	IgniteCache cache = grid(igniteIdx).cache(CACHE_PREFIX + cacheIdx);

	CacheMetrics metrics = cache.metrics();

	String cacheInfo = "igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx + " ";

	assertEquals(cacheInfo + " CacheSize", cacheSize, metrics.getCacheSize());
	assertEquals(cacheInfo + " offHeapEntriesCnt", offHeapEntriesCnt,
	metrics.getOffHeapEntriesCount());
	assertEquals(cacheInfo + " offHeapBackupEntriesCnt", offHeapBackupEntriesCnt,
	metrics.getOffHeapBackupEntriesCount());
	assertEquals(cacheInfo + " offHeapPrimaryEntriesCnt", offHeapPrimaryEntriesCnt,
	metrics.getOffHeapPrimaryEntriesCount());

	if (!MvccFeatureChecker.forcedMvcc()) // Onheap cache is not supported in Mvcc mode.
	assertEquals(cacheInfo + " heapEntriesCnt", heapEntriesCnt, metrics.getHeapEntriesCount());

	assertEquals(cacheInfo + " size", cacheSize, metrics.getSize());
	assertEquals(cacheInfo + " keySize", cacheSize, metrics.getKeySize());
	assertEquals(cacheInfo + " isEmpty", cacheSize == 0, metrics.isEmpty());

	metrics = cache.localMetrics();

	cacheSizeSum += metrics.getCacheSize();
	offHeapEntriesCntSum += metrics.getOffHeapEntriesCount();
	offHeapPrimaryEntriesCntSum += metrics.getOffHeapPrimaryEntriesCount();
	offHeapBackupEntriesCntSum += metrics.getOffHeapBackupEntriesCount();
	heapEntriesCntSum += metrics.getHeapEntriesCount();
	isEmptySum = isEmptySum && metrics.isEmpty();
	}

	String cacheInfo = "cacheIdx=" + cacheIdx + " check sum";

	assertEquals(cacheInfo + " CacheSize", cacheSize, cacheSizeSum);
	assertEquals(cacheInfo + " offHeapEntriesCnt", offHeapEntriesCnt, offHeapEntriesCntSum);
	assertEquals(cacheInfo + " offHeapBackupEntriesCnt", offHeapBackupEntriesCnt, offHeapBackupEntriesCntSum);
	assertEquals(cacheInfo + " offHeapPrimaryEntriesCnt", offHeapPrimaryEntriesCnt, offHeapPrimaryEntriesCntSum);

	if (!MvccFeatureChecker.forcedMvcc()) // Onheap cache is not supported in Mvcc mode.
	assertEquals(cacheInfo + " heapEntriesCnt", heapEntriesCnt, heapEntriesCntSum);

	assertEquals(cacheInfo + " isEmpty", cacheSize == 0, isEmptySum);
	}
	}