fix after review

Author: zstan

modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryAbstractDistributedJoinSelfTest.java

@@ -19,14 +19,13 @@
 
 import java.io.Serializable;
 import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
 import org.apache.ignite.cache.query.annotations.QuerySqlField;
-import org.apache.ignite.cache.query.annotations.QuerySqlFunction;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.processors.cache.index.AbstractIndexingCommonTest;
 import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.GridTestUtils;
 
 import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
 import static org.apache.ignite.cache.CacheMode.PARTITIONED;
@@ -62,10 +61,10 @@ public class IgniteCacheQueryAbstractDistributedJoinSelfTest extends AbstractInd
         "order by co._key, pr._key ";
 
     /** */
-    protected static final String QRY_LONG = "select pe.id, co.id, pr._key\n" +
+    protected static final String QRY_LONG = "select id1 from (select pe.id as id1, co.id, pr._key\n" +
         "from \"pe\".Person pe, \"pr\".Product pr, \"co\".Company co, \"pu\".Purchase pu\n" +
-        "where pe._key = pu.personId and pu.productId = pr._key and pr.companyId = co._key \n" +
-        "order by pe.id desc";
+        "where pe._key = pu.personId and pu.productId = pr._key and pr.companyId = co._key\n" +
+        "order by pe.id) where id1 > sleep(10)";
 
     /** */
     protected static final int GRID_CNT = 2;
@@ -74,7 +73,7 @@ public class IgniteCacheQueryAbstractDistributedJoinSelfTest extends AbstractInd
     private static final int PERS_CNT = 600;
 
     /** */
-    private static final int PURCHASE_CNT = 24_000;
+    private static final int PURCHASE_CNT = 6_000;
 
     /** */
     private static final int COMPANY_CNT = 25;
@@ -100,6 +99,8 @@ public class IgniteCacheQueryAbstractDistributedJoinSelfTest extends AbstractInd
             cc.setAtomicityMode(TRANSACTIONAL);
             cc.setRebalanceMode(SYNC);
             cc.setLongQueryWarningTimeout(15_000);
+            cc.setAffinity(new RendezvousAffinityFunction(false, 60));
+            cc.setSqlFunctionClasses(GridTestUtils.SqlTestFunctions.class);
 
             switch (name) {
                 case "pe":
@@ -252,20 +253,4 @@ protected static class Product implements Serializable {
             this.companyId = companyId;
         }
     }
-
-    /** */
-    public static class Functions {
-        /** */
-        @QuerySqlFunction
-        public static int sleep() {
-            try {
-                U.sleep(1_000);
-            }
-            catch (IgniteInterruptedCheckedException ignored) {
-                // No-op.
-            }
-
-            return 0;
-        }
-    }
 }

modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryStopOnCancelOrTimeoutDistributedJoinSelfTest.java

@@ -25,16 +25,20 @@
 import javax.cache.CacheException;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
 import org.apache.ignite.cache.query.QueryCancelledException;
 import org.apache.ignite.cache.query.QueryCursor;
 import org.apache.ignite.cache.query.SqlFieldsQuery;
 import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.processors.GridProcessor;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.X;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.junit.Test;
 
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
 /**
  * Test for cancel of query containing distributed joins.
  */
@@ -90,7 +94,7 @@ public void testTimeout4() throws Exception {
     /** */
     private void testQueryCancel(Ignite ignite, String cacheName, String sql, int timeoutUnits, TimeUnit timeUnit,
                            boolean timeout, boolean checkCanceled) throws Exception {
-        SqlFieldsQuery qry = new SqlFieldsQuery(sql).setDistributedJoins(true);;
+        SqlFieldsQuery qry = new SqlFieldsQuery(sql).setDistributedJoins(true);
 
         IgniteCache<Object, Object> cache = ignite.cache(cacheName);
 
@@ -107,44 +111,36 @@ private void testQueryCancel(Ignite ignite, String cacheName, String sql, int ti
         }
 
         try (QueryCursor<List<?>> ignored = cursor) {
-            int resSize = 0;
-            for (List<?> ignored1 : cursor) {
-                ++resSize;
-            }
+            int resSize = F.size(cursor.iterator());
 
             if (checkCanceled)
                 fail("Query not canceled, result size=" + resSize);
         }
-        catch (CacheException ex) {
-            log().error("Got expected exception", ex);
+        catch (CacheException | IgniteException ex) {
+            log().error("Got exception", ex);
 
             assertNotNull("Must throw correct exception", X.cause(ex, QueryCancelledException.class));
         }
 
-        // Give some time to clean up.
-        Thread.sleep(TimeUnit.MILLISECONDS.convert(timeoutUnits, timeUnit) + 1_000);
-
         checkCleanState();
     }
 
     /**
      * Validates clean state on all participating nodes after query cancellation.
      */
-    private void checkCleanState() {
+    private void checkCleanState() throws IgniteInterruptedCheckedException {
         for (int i = 0; i < GRID_CNT; i++) {
             IgniteEx grid = grid(i);
 
             // Validate everything was cleaned up.
-            ConcurrentMap<UUID, ?> map = U.field(((IgniteH2Indexing)U.field((GridProcessor)U.field(
-                    grid.context(), "qryProc"), "idx")).mapQueryExecutor(), "qryRess");
-
-            String msg = "Map executor state is not cleared";
+            ConcurrentMap<UUID, ?> map = U.field(
+                ((IgniteH2Indexing)grid.context().query().getIndexing()).mapQueryExecutor(), "qryRess");
 
             // TODO FIXME Current implementation leaves map entry for each node that's ever executed a query.
             for (Object result : map.values()) {
                 Map<Long, ?> m = U.field(result, "res");
 
-                assertEquals(msg, 0, m.size());
+                assertTrue("Map executor state is not cleared", waitForCondition(m::isEmpty, 1_000L));
             }
         }
     }