You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ignite.apache.org by "Alexei Scherbakov (JIRA)" <ji...@apache.org> on 2019/04/25 10:17:00 UTC
[jira] [Updated] (IGNITE-11804) Assertion error on affinity
initialization
[ https://issues.apache.org/jira/browse/IGNITE-11804?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Alexei Scherbakov updated IGNITE-11804:
---------------------------------------
Summary: Assertion error on affinity initialization (was: Assertion error during assignment calculation)
> Assertion error on affinity initialization
> ------------------------------------------
>
> Key: IGNITE-11804
> URL: https://issues.apache.org/jira/browse/IGNITE-11804
> Project: Ignite
> Issue Type: Bug
> Reporter: Alexei Scherbakov
> Priority: Major
>
> Reproducer (needs some cleanup)
> {noformat}
> /*
> * 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.transactions;
> import java.util.ArrayList;
> import java.util.List;
> import java.util.concurrent.atomic.AtomicReference;
> import org.apache.ignite.Ignite;
> 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.GridCacheContext;
> import org.apache.ignite.internal.processors.cache.IgniteCacheProxyImpl;
> import org.apache.ignite.internal.processors.cache.persistence.db.wal.IgniteWalRebalanceTest;
> import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
> import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
> import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
> import org.apache.ignite.transactions.Transaction;
> import org.junit.Test;
> import static java.util.concurrent.TimeUnit.DAYS;
> import static java.util.concurrent.TimeUnit.MILLISECONDS;
> import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
> import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
> import static org.apache.ignite.configuration.WALMode.LOG_ONLY;
> import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
> import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
> /**
> * Test framework for ordering transaction's prepares and commits by intercepting messages and releasing then
> * in user defined order.
> */
> public class TxPartitionCounterStateAbstractTest extends GridCommonAbstractTest {
> /** IP finder. */
> private static final TcpDiscoveryVmIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
> /** */
> private static final int MB = 1024 * 1024;
> /** */
> protected int backups;
> /** */
> public static final int TEST_TIMEOUT = 30_000;
> public static final String DEFAULT_CACHE_NAME_2 = DEFAULT_CACHE_NAME + "2";
> /** */
> private AtomicReference<Throwable> testFailed = new AtomicReference<>();
> /** Number of keys to preload before txs to enable historical rebalance. */
> protected static final int PRELOAD_KEYS_CNT = 1;
> /** */
> protected static final String CLIENT_GRID_NAME = "client";
> /** */
> protected static final int PARTS_CNT = 32;
> /** {@inheritDoc} */
> @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
> IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
> cfg.setConsistentId("node" + igniteInstanceName);
> cfg.setFailureHandler(new StopNodeFailureHandler());
> cfg.setRebalanceThreadPoolSize(4); // Necessary to reproduce some issues.
> ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(IP_FINDER);
> // TODO set this only for historical rebalance tests.
> cfg.setCommunicationSpi(new IgniteWalRebalanceTest.WalRebalanceCheckingCommunicationSpi());
> boolean client = igniteInstanceName.startsWith(CLIENT_GRID_NAME);
> cfg.setClientMode(client);
> cfg.setDataStorageConfiguration(new DataStorageConfiguration().
> setWalHistorySize(1000).
> setWalSegmentSize(8 * MB).setWalMode(LOG_ONLY).setPageSize(1024).
> setCheckpointFrequency(MILLISECONDS.convert(365, DAYS)).
> setDefaultDataRegionConfiguration(new DataRegionConfiguration().setPersistenceEnabled(true).
> setInitialSize(100 * MB).setMaxSize(100 * MB)));
> if (!client)
> cfg.setCacheConfiguration(cacheConfiguration(DEFAULT_CACHE_NAME));
> return cfg;
> }
> /**
> * @param name Name.
> */
> protected CacheConfiguration cacheConfiguration(String name) {
> CacheConfiguration ccfg = new CacheConfiguration(name);
> ccfg.setAtomicityMode(TRANSACTIONAL);
> ccfg.setBackups(backups);
> ccfg.setWriteSynchronizationMode(FULL_SYNC);
> ccfg.setOnheapCacheEnabled(false);
> ccfg.setAffinity(new RendezvousAffinityFunction(false, PARTS_CNT));
> ccfg.setGroupName("test");
> return ccfg;
> }
> /** {@inheritDoc} */
> @Override protected void beforeTest() throws Exception {
> super.beforeTest();
> cleanPersistenceDir();
> }
> /** {@inheritDoc} */
> @Override protected void afterTest() throws Exception {
> super.afterTest();
> stopAllGrids();
> cleanPersistenceDir();
> }
> @Test
> public void testPut() throws Exception {
> backups = 2;
> Ignite crd = startGridsMultiThreaded(3);
> IgniteEx client = startGrid("client");
> assertNotNull(client.createCache(cacheConfiguration(DEFAULT_CACHE_NAME_2)));
> try(Transaction tx = client.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
> List<Integer> keys0 = partitionKeys(client.cache(DEFAULT_CACHE_NAME), 1, 10, 0);
> List<Integer> keys2 = partitionKeys(client.cache(DEFAULT_CACHE_NAME_2), 9, 10, 0);
> for (Integer k : keys0)
> client.cache(DEFAULT_CACHE_NAME).put(k, k);
> for (Integer k : keys2)
> client.cache(DEFAULT_CACHE_NAME_2).put(k, k);
> tx.commit();
> }
> }
> /**
> * @param cache Cache.
> * @param part Partition.
> * @param cnt Count.
> * @param skipCnt Skip keys from start.
> * @return List of keys for partition.
> */
> protected List<Integer> partitionKeys(IgniteCache<?, ?> cache, int part, int cnt, int skipCnt) {
> IgniteCacheProxyImpl proxy = cache.unwrap(IgniteCacheProxyImpl.class);
> GridCacheContext<?, ?> cctx = proxy.context();
> int k = 0, c = 0, skip0 = 0;
> List<Integer> keys = new ArrayList<>(cnt);
> while(c < cnt) {
> if (cctx.affinity().partition(k) == part) {
> if (skip0 < skipCnt) {
> k++;
> skip0++;
> continue;
> }
> c++;
> keys.add(k);
> }
> k++;
> }
> return keys;
> }
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)