ignite-446

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java

@@ -471,9 +471,8 @@ private void leaveBusy() {
     /**
      * @return Exchange futures.
      */
-    @SuppressWarnings( {"unchecked", "RedundantCast"})
-    public List<IgniteInternalFuture<?>> exchangeFutures() {
-        return (List<IgniteInternalFuture<?>>)(List)exchFuts.values();
+    public List<GridDhtPartitionsExchangeFuture> exchangeFutures() {
+        return exchFuts.values();
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java

@@ -528,6 +528,17 @@ void map() {
             topVer = tx.topologyVersionSnapshot();
 
         if (topVer != null) {
+            for (GridDhtTopologyFuture fut : cctx.shared().exchange().exchangeFutures()){
+                if (fut.topologyVersion().equals(topVer)){
+                    if (!fut.isCacheTopologyValid(cctx)) {
+                        onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
+                            cctx.name()));
+
+                        return;
+                    }
+                }
+            }
+
             // Continue mapping on the same topology version as it was before.
             this.topVer.compareAndSet(null, topVer);
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java

@@ -660,6 +660,17 @@ void map() {
             topVer = tx.topologyVersionSnapshot();
 
         if (topVer != null) {
+            for (GridDhtTopologyFuture fut : cctx.shared().exchange().exchangeFutures()){
+                if (fut.topologyVersion().equals(topVer)){
+                    if (!fut.isCacheTopologyValid(cctx)) {
+                        onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
+                            cctx.name()));
+
+                        return;
+                    }
+                }
+            }
+
             // Continue mapping on the same topology version as it was before.
             this.topVer.compareAndSet(null, topVer);
 

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractCacheTest.java

@@ -116,6 +116,7 @@ protected void putValid(String cacheName) {
 
     /**
      * Commits with error.
+     *
      * @param tx transaction.
      */
     protected void commitFailed(Transaction tx) {
@@ -129,18 +130,20 @@ protected void commitFailed(Transaction tx) {
     }
 
     /**
-     * Removes key-value
+     * Removes key-value.
      *
      * @param cacheName cache name.
      */
     public void remove(String cacheName) {
+        assert grid(0).cache(cacheName).get(KEY_VALUE) != null;
+
         grid(0).cache(cacheName).remove(KEY_VALUE);
     }
 
     /**
      * Asserts that cache doesn't contains key.
      *
-     * @param cacheName
+     * @param cacheName cache name.
      */
     public void assertEmpty(String cacheName) {
         assert grid(0).cache(cacheName).get(KEY_VALUE) == null;

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheTest.java

@@ -59,15 +59,15 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
 
         assertEmpty(null); // rolled back
         assertEmpty(CACHE_NAME_1); // rolled back
+        assertEmpty(CACHE_NAME_2); // rolled back
 
-//        try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
-//            putValid(null);
-//            putInvalid(CACHE_NAME_1);
-//        }
-//
-//        assertEmpty(null); // rolled back
-//        assertEmpty(CACHE_NAME_1); // rolled back
+        try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+            putValid(null);
+            putInvalid(CACHE_NAME_1);
+        }
 
+        assertEmpty(null); // rolled back
+        assertEmpty(CACHE_NAME_1); // rolled back
 
         startGrid(1);
 
@@ -87,6 +87,14 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
 
         startGrid(2);
 
+        try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+            putValid(null);
+            putInvalid(CACHE_NAME_1);
+        }
+
+        assertEmpty(null); // rolled back
+        assertEmpty(CACHE_NAME_1); // rolled back
+
         try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
             putValid(CACHE_NAME_1);
             commitFailed(tx);
@@ -96,5 +104,22 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
             putInvalid(CACHE_NAME_1);
         }
 
+        try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+            putValid(null);
+            putValid(CACHE_NAME_2);
+            tx.commit();
+        }
+
+        remove(null);
+        remove(CACHE_NAME_2);
+
+        try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+            putValid(null);
+            putValid(CACHE_NAME_2);
+            tx.commit();
+        }
+
+        remove(null);
+        remove(CACHE_NAME_2);
     }
 }