This is an automated email from the ASF dual-hosted git repository. irakov 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 31506f0 IGNITE-12392 Faster transaction rolled back when one of backup node failed - Fixes #7072. 31506f0 is described below commit 31506f0600c433e8873d3f44590e0c34f91a9122 Author: Kirill Tkalenko <tkalkir...@yandex.ru> AuthorDate: Thu Nov 28 21:15:08 2019 +0300 IGNITE-12392 Faster transaction rolled back when one of backup node failed - Fixes #7072. Signed-off-by: Ivan Rakov <ira...@apache.org> --- .../distributed/dht/GridDhtTxFinishFuture.java | 27 +- .../GridCacheFastNodeLeftForTransactionTest.java | 394 +++++++++++++++++++++ .../ignite/testsuites/IgniteCacheTestSuite2.java | 2 + 3 files changed, 414 insertions(+), 9 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java index dc29ebe..00d1588 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java @@ -48,6 +48,7 @@ import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteUuid; +import static java.util.Objects.isNull; import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC; import static org.apache.ignite.transactions.TransactionState.COMMITTING; @@ -467,18 +468,26 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCacheCompoundIdentity req.writeVersion(tx.writeVersion() != null ? tx.writeVersion() : tx.xidVersion()); try { - cctx.io().send(n, req, tx.ioPolicy()); + if (isNull(cctx.discovery().getAlive(n.id()))) { + log.error("Unable to send message (node left topology): " + n); - if (msgLog.isDebugEnabled()) { - msgLog.debug("DHT finish fut, sent request dht [txId=" + tx.nearXidVersion() + - ", dhtTxId=" + tx.xidVersion() + - ", node=" + n.id() + ']'); + fut.onNodeLeft(new ClusterTopologyCheckedException("Node left grid while sending message to: " + + n.id())); } + else { + cctx.io().send(n, req, tx.ioPolicy()); - if (sync) - res = true; - else - fut.onDone(); + if (msgLog.isDebugEnabled()) { + msgLog.debug("DHT finish fut, sent request dht [txId=" + tx.nearXidVersion() + + ", dhtTxId=" + tx.xidVersion() + + ", node=" + n.id() + ']'); + } + + if (sync) + res = true; + else + fut.onDone(); + } } catch (IgniteCheckedException e) { // Fail the whole thing. diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheFastNodeLeftForTransactionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheFastNodeLeftForTransactionTest.java new file mode 100644 index 0000000..6bccbd2 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheFastNodeLeftForTransactionTest.java @@ -0,0 +1,394 @@ +/* + * 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.local; + +import java.io.ByteArrayOutputStream; +import java.io.OutputStream; +import java.io.PrintStream; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; +import java.util.logging.Formatter; +import java.util.logging.Handler; +import java.util.logging.LogRecord; +import java.util.logging.Logger; +import java.util.logging.StreamHandler; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.ConnectorConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.commandline.CommandHandler; +import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishFuture; +import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.GridAbstractTest; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.apache.ignite.transactions.Transaction; +import org.junit.Test; + +import static java.util.Arrays.asList; +import static java.util.Arrays.stream; +import static java.util.Objects.nonNull; +import static java.util.function.Function.identity; +import static java.util.stream.Collectors.toList; +import static java.util.stream.Collectors.toMap; +import static java.util.stream.IntStream.range; +import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC; +import static org.apache.ignite.internal.commandline.CommandHandler.initLogger; +import static org.apache.ignite.testframework.GridTestUtils.assertContains; +import static org.apache.ignite.testframework.GridTestUtils.getFieldValue; +import static org.apache.ignite.testframework.GridTestUtils.setFieldValue; +import static org.apache.ignite.testframework.LogListener.matches; + +/** + * Class for testing fast node left during transaction for cache. + */ +public class GridCacheFastNodeLeftForTransactionTest extends GridCommonAbstractTest { + /** Number of nodes. */ + private static final int NODES = 4; + + /** Number of transactions. */ + private static final int TX_COUNT = 20; + + /** Logger for listen log messages. */ + private static ListeningTestLogger listeningLog; + + /** Creating a client node. */ + private boolean clientNode; + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + super.beforeTestsStarted(); + + listeningLog = new ListeningTestLogger(false, GridAbstractTest.log); + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + /*To listen the logs of future in current tests, since the log in the + futures is static and is not reset when tests are launched.*/ + setFieldValue(GridDhtTxFinishFuture.class, "log", null); + ((AtomicReference<IgniteLogger>)getFieldValue(GridDhtTxFinishFuture.class, "logRef")).set(null); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + + listeningLog.clearListeners(); + + super.afterTest(); + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + return super.getConfiguration(igniteInstanceName) + .setClientMode(clientNode) + .setCacheConfiguration(createCacheConfigs()) + .setGridLogger(listeningLog) + .setConnectorConfiguration(new ConnectorConfiguration()); + } + + /** + * Test transaction rollback when one of the nodes drops out. + * + * @throws Exception If failed. + */ + @Test + public void testRollbackTransactions() throws Exception { + int txCnt = TX_COUNT; + + int nodes = NODES; + + IgniteEx crd = createCluster(nodes); + + for (CacheConfiguration cacheConfig : createCacheConfigs()) { + String cacheName = cacheConfig.getName(); + + IgniteCache<Object, Object> cache = crd.cache(cacheName); + + List<Integer> keys = primaryKeys(cache, txCnt); + + Map<Integer, Integer> cacheValues = range(0, txCnt / 2).boxed().collect(toMap(keys::get, identity())); + + cache.putAll(cacheValues); + + Collection<Transaction> txs = createTxs( + grid(nodes), + cacheName, + range(txCnt / 2, txCnt).mapToObj(keys::get).collect(toList()) + ); + + int stoppedNodeId = 2; + + stopGrid(stoppedNodeId); + + LogListener logLsnr = newLogListener(); + + listeningLog.registerListener(logLsnr); + + for (Transaction tx : txs) + tx.rollback(); + + awaitPartitionMapExchange(); + + check(cacheValues, cacheName, logLsnr, stoppedNodeId); + } + } + + /** + * Test for rollback transactions when one of the nodes drops out, + * with operations performed on keys outside the transaction. + * + * @throws Exception If failed. + */ + @Test + public void testRollbackTransactionsWithKeyOperationOutsideThem() throws Exception { + int txCnt = TX_COUNT; + + int nodes = NODES; + + IgniteEx crd = createCluster(nodes); + + for (CacheConfiguration cacheConfig : createCacheConfigs()) { + String cacheName = cacheConfig.getName(); + + IgniteCache<Object, Object> cache = crd.cache(cacheName); + + List<Integer> keys = primaryKeys(cache, txCnt); + + Map<Integer, Integer> cacheValues = range(0, txCnt / 2).boxed().collect(toMap(keys::get, identity())); + + cache.putAll(cacheValues); + + List<Integer> txKeys = range(txCnt / 2, txCnt).mapToObj(keys::get).collect(toList()); + + IgniteEx clientNode = grid(nodes); + + Collection<Transaction> txs = createTxs(clientNode, cacheName, txKeys); + + int stoppedNodeId = 2; + + stopGrid(stoppedNodeId); + + CountDownLatch latch = new CountDownLatch(1); + + GridTestUtils.runAsync(() -> { + latch.countDown(); + + IgniteCache<Object, Object> clientCache = clientNode.cache(DEFAULT_CACHE_NAME); + + txKeys.forEach(clientCache::get); + }); + + LogListener logLsnr = newLogListener(); + + listeningLog.registerListener(logLsnr); + + latch.await(); + + for (Transaction tx : txs) + tx.rollback(); + + awaitPartitionMapExchange(); + + check(cacheValues, cacheName, logLsnr, stoppedNodeId); + } + } + + /** + * Checking the contents of the cache after rollback transactions, + * with restarting the stopped node with using "idle_verify". + * + * @param cacheValues Expected cache contents. + * @param cacheName Cache name. + * @param logLsnr LogListener. + * @param stoppedNodeId ID of the stopped node. + * @throws Exception If failed. + */ + private void check( + Map<Integer, Integer> cacheValues, + String cacheName, + LogListener logLsnr, + int stoppedNodeId + ) throws Exception { + assert nonNull(cacheValues); + assert nonNull(cacheName); + assert nonNull(logLsnr); + + checkCacheData(cacheValues, cacheName); + + assertTrue(logLsnr.check()); + + clientNode = false; + + startGrid(stoppedNodeId); + + awaitPartitionMapExchange(); + + checkCacheData(cacheValues, cacheName); + + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + + PrintStream sysOut = System.out; + + try(PrintStream out = new PrintStream(baos)) { + System.setOut(out); + + Logger cmdLog = createTestLogger(baos); + CommandHandler cmdHnd = new CommandHandler(cmdLog); + + cmdHnd.execute(asList("--cache", "idle_verify")); + + stream(cmdLog.getHandlers()).forEach(Handler::flush); + + assertContains(listeningLog, baos.toString(), "no conflicts have been found"); + } + finally { + System.setOut(sysOut); + } + } + + /** + * Creating a logger for a CommandHandler. + * + * @param outputStream Stream for recording the result of a command. + * @return Logger. + */ + private Logger createTestLogger(OutputStream outputStream) { + assert nonNull(outputStream); + + Logger log = initLogger(null); + + log.addHandler(new StreamHandler(outputStream, new Formatter() { + @Override public String format(LogRecord record) { + return record.getMessage() + "\n"; + } + })); + + return log; + } + + /** + * Creating a cluster. + * + * @param nodes Number of server nodes, plus one client. + * @throws Exception If failed. + */ + private IgniteEx createCluster(int nodes) throws Exception { + clientNode = false; + + IgniteEx crd = startGrids(nodes); + + clientNode = true; + + startGrid(nodes); + + awaitPartitionMapExchange(); + + return crd; + } + + /** + * Transaction creation. + * + * @param node Node. + * @param cacheName Cache name. + * @param keys Keys. + * @return Transactions. + * @throws Exception If failed. + */ + private Collection<Transaction> createTxs( + IgniteEx node, + String cacheName, + Collection<Integer> keys + ) throws Exception { + assert nonNull(node); + assert nonNull(cacheName); + assert nonNull(keys); + + IgniteCache<Object, Object> cache = node.cache(cacheName); + + Collection<Transaction> txs = new ArrayList<>(); + + for (Integer key : keys) { + Transaction tx = node.transactions().txStart(); + + cache.put(key, key + 10); + + ((TransactionProxyImpl)tx).tx().prepare(true); + + txs.add(tx); + } + + return txs; + } + + /** + * Creating an instance of LogListener to find an exception + * "Unable to send message (node left topology):". + * + * @return LogListener. + */ + private LogListener newLogListener() { + return matches("Unable to send message (node left topology):").build(); + } + + /** + * Creating a cache configurations. + * + * @return Cache configurations. + */ + private CacheConfiguration[] createCacheConfigs() { + return new CacheConfiguration[] { + createCacheConfig(DEFAULT_CACHE_NAME + "_0", FULL_SYNC), + createCacheConfig(DEFAULT_CACHE_NAME + "_1", PRIMARY_SYNC) + }; + } + + /** + * Creating a cache configuration. + * + * @param cacheName Cache name. + * @param syncMode Sync mode. + * @return Cache configuration. + */ + private CacheConfiguration createCacheConfig(String cacheName, CacheWriteSynchronizationMode syncMode) { + assert nonNull(cacheName); + assert nonNull(syncMode); + + return new CacheConfiguration(cacheName) + .setAtomicityMode(TRANSACTIONAL) + .setBackups(2) + .setAffinity(new RendezvousAffinityFunction(false, 10)) + .setWriteSynchronizationMode(syncMode); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java index 799ff80..ee42927 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java @@ -162,6 +162,7 @@ import org.apache.ignite.internal.processors.cache.local.GridCacheLocalIsolatedN import org.apache.ignite.internal.processors.cache.local.GridCacheLocalLoadAllSelfTest; import org.apache.ignite.internal.processors.cache.local.GridCacheLocalLockSelfTest; import org.apache.ignite.internal.processors.cache.local.GridCacheLocalMultithreadedSelfTest; +import org.apache.ignite.internal.processors.cache.local.GridCacheFastNodeLeftForTransactionTest; import org.apache.ignite.internal.processors.cache.local.GridCacheLocalTxMultiThreadedSelfTest; import org.apache.ignite.internal.processors.cache.local.GridCacheLocalTxReadTest; import org.apache.ignite.internal.processors.cache.local.GridCacheLocalTxSingleThreadedSelfTest; @@ -209,6 +210,7 @@ public class IgniteCacheTestSuite2 { GridTestUtils.addTestIfNeeded(suite, GridCacheLocalEvictionEventSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCacheLocalTxMultiThreadedSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCacheLocalIsolatedNodesSelfTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, GridCacheFastNodeLeftForTransactionTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCacheTransformEventSelfTest.class, ignoredTests);