ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a.@apache.org
Subject [1/2] ignite git commit: Squashed commit of the following:
Date Tue, 26 Jan 2016 19:30:43 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-2224-2 c55cc975d -> 920089ce4


http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 6130ead..942f0a7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -477,7 +477,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
     /** {@inheritDoc} */
 
     @SuppressWarnings("unchecked")
-    @Override @Nullable public V get(K key, boolean deserializeBinary) throws IgniteCheckedException
{
+    @Override @Nullable public V get(K key, boolean deserializeBinary, boolean needVer) throws
IgniteCheckedException {
         String taskName = ctx.kernalContext().job().currentTaskName();
 
         Map<K, V> m = getAllInternal(Collections.singleton(key),
@@ -485,7 +485,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
             ctx.readThrough(),
             taskName,
             deserializeBinary,
-            false);
+            false,
+            needVer);
 
         assert m.isEmpty() || m.size() == 1 : m.size();
 
@@ -494,7 +495,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
 
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
-    @Override public final Map<K, V> getAll(Collection<? extends K> keys, boolean
deserializeBinary)
+    @Override public final Map<K, V> getAll(Collection<? extends K> keys, boolean
deserializeBinary, boolean needVer)
         throws IgniteCheckedException {
         A.notNull(keys, "keys");
 
@@ -505,7 +506,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
             ctx.readThrough(),
             taskName,
             deserializeBinary,
-            false);
+            false,
+            needVer);
     }
 
 
@@ -519,7 +521,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
         final String taskName,
         final boolean deserializeBinary,
         final boolean skipVals,
-        boolean canRemap
+        boolean canRemap,
+        final boolean needVer
     ) {
         A.notNull(keys, "keys");
 
@@ -528,7 +531,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
 
         return asyncOp(new Callable<Map<K, V>>() {
             @Override public Map<K, V> call() throws Exception {
-                return getAllInternal(keys, swapOrOffheap, storeEnabled, taskName, deserializeBinary,
skipVals);
+                return getAllInternal(keys, swapOrOffheap, storeEnabled, taskName, deserializeBinary,
skipVals, needVer);
             }
         });
     }
@@ -542,6 +545,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
      * @param taskName Task name.
      * @param deserializeBinary Deserialize binary .
      * @param skipVals Skip value flag.
+     * @param needVer Need version.
      * @return Key-value map.
      * @throws IgniteCheckedException If failed.
      */
@@ -551,7 +555,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
         boolean storeEnabled,
         String taskName,
         boolean deserializeBinary,
-        boolean skipVals
+        boolean skipVals,
+        boolean needVer
     ) throws IgniteCheckedException {
         ctx.checkSecurity(SecurityPermission.CACHE_READ);
 
@@ -584,24 +589,66 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
                     entry = swapOrOffheap ? entryEx(cacheKey) : peekEx(cacheKey);
 
                     if (entry != null) {
-                        CacheObject v = entry.innerGet(null,
-                            /*swap*/swapOrOffheap,
-                            /*read-through*/false,
-                            /*fail-fast*/false,
-                            /*unmarshal*/true,
-                            /**update-metrics*/true,
-                            /**event*/!skipVals,
-                            /**temporary*/false,
-                            subjId,
-                            null,
-                            taskName,
-                            expiry,
-                            !deserializeBinary);
 
-                        if (v != null)
-                            ctx.addResult(vals, cacheKey, v, skipVals, false, deserializeBinary,
true);
-                        else
-                            success = false;
+                        CacheObject v = null;
+                        GridCacheVersion ver = null;
+
+                        if (needVer) {
+                            T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
+                                null,
+                                /*swap*/swapOrOffheap,
+                                /*unmarshal*/true,
+                                /**update-metrics*/false,
+                                /*event*/!skipVals,
+                                subjId,
+                                null,
+                                taskName,
+                                expiry,
+                                !deserializeBinary);
+
+                            if (res != null) {
+                                v = res.get1();
+                                ver = res.get2();
+
+                                ctx.addResult(
+                                    vals,
+                                    cacheKey,
+                                    v,
+                                    skipVals,
+                                    false,
+                                    deserializeBinary,
+                                    true,
+                                    ver);
+                            }else
+                                success = false;
+                        }
+                        else {
+                            v = entry.innerGet(null,
+                                /*swap*/swapOrOffheap,
+                                /*read-through*/false,
+                                /*fail-fast*/false,
+                                /*unmarshal*/true,
+                                /**update-metrics*/true,
+                                /**event*/!skipVals,
+                                /**temporary*/false,
+                                subjId,
+                                null,
+                                taskName,
+                                expiry,
+                                !deserializeBinary);
+
+                            if (v != null) {
+                                ctx.addResult(vals,
+                                    cacheKey,
+                                    v,
+                                    skipVals,
+                                    false,
+                                    deserializeBinary,
+                                    true);
+                            }
+                            else
+                                success = false;
+                        }
                     }
                     else {
                         if (!storeEnabled && configuration().isStatisticsEnabled()
&& !skipVals)
@@ -638,7 +685,8 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
             /*force primary*/false,
             expiry,
             skipVals,
-            /*can remap*/true).get();
+            /*can remap*/true,
+            needVer).get();
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 926eaf2..6422305 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1342,7 +1342,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
 
     /**
      * Checks if there is a cached or swapped value for
-     * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean)}
method.
+     * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean,
boolean)} method.
      *
      * @param cacheCtx Cache context.
      * @param keys Key to enlist.
@@ -1368,7 +1368,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
         boolean deserializeBinary,
         boolean skipVals,
         boolean keepCacheObjects,
-        boolean skipStore
+        boolean skipStore,
+        final boolean needVer
     ) throws IgniteCheckedException {
         assert !F.isEmpty(keys);
         assert keysCnt == keys.size();
@@ -1381,7 +1382,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
 
         AffinityTopologyVersion topVer = topologyVersion();
 
-        boolean needReadVer = serializable() && optimistic();
+        boolean needReadVer = (serializable() && optimistic()) || needVer;
 
         // In this loop we cover only read-committed or optimistic transactions.
         // Transactions that are pessimistic and not read-committed are covered
@@ -1403,31 +1404,67 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
implements Ig
                     if (!F.isEmpty(txEntry.entryProcessors()))
                         val = txEntry.applyEntryProcessors(val);
 
-                    if (val != null)
-                        cacheCtx.addResult(map, key, val, skipVals, keepCacheObjects, deserializeBinary,
false);
+                    if (val != null) {
+                        GridCacheVersion ver = null;
+
+                        if (needVer) {
+                            try {
+                                ver = txEntry.cached().version();
+                            }
+                            catch (GridCacheEntryRemovedException ignored) {
+                                if (log.isDebugEnabled())
+                                    log.debug("Got removed entry in transaction getAllAsync(..)"
+ key);
+                            }
+                        }
+
+                        cacheCtx.addResult(map, key, val, skipVals, keepCacheObjects, deserializeBinary,
false, ver);
+                    }
                 }
                 else {
                     assert txEntry.op() == TRANSFORM;
 
                     while (true) {
                         try {
+                            GridCacheVersion readVer = null;
+
                             Object transformClo =
-                                (txEntry.op() == TRANSFORM  && cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ))
?
+                                (txEntry.op() == TRANSFORM &&
+                                    cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ))
?
                                     F.first(txEntry.entryProcessors()) : null;
 
-                            val = txEntry.cached().innerGet(this,
-                                /*swap*/true,
-                                /*read-through*/false,
-                                /*fail fast*/true,
-                                /*unmarshal*/true,
-                                /*metrics*/true,
-                                /*event*/!skipVals,
-                                /*temporary*/false,
-                                CU.subjectId(this, cctx),
-                                transformClo,
-                                resolveTaskName(),
-                                null,
-                                txEntry.keepBinary());
+                            if (needVer) {
+                                T2<CacheObject, GridCacheVersion> res = txEntry.cached().innerGetVersioned(
+                                    this,
+                                    /*swap*/true,
+                                    /*unmarshal*/true,
+                                    /*update-metrics*/true,
+                                    /*event*/!skipVals,
+                                    CU.subjectId(this, cctx),
+                                    transformClo,
+                                    resolveTaskName(),
+                                    null,
+                                    txEntry.keepBinary());
+
+                                if (res != null) {
+                                    val = res.get1();
+                                    readVer = res.get2();
+                                }
+                            }
+                            else {
+                                val = txEntry.cached().innerGet(this,
+                                    /*swap*/true,
+                                    /*read-through*/false,
+                                    /*fail fast*/true,
+                                    /*unmarshal*/true,
+                                    /*metrics*/true,
+                                    /*event*/!skipVals,
+                                    /*temporary*/false,
+                                    CU.subjectId(this, cctx),
+                                    transformClo,
+                                    resolveTaskName(),
+                                    null,
+                                    txEntry.keepBinary());
+                            }
 
                             if (val != null) {
                                 if (!readCommitted() && !skipVals)
@@ -1442,7 +1479,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
                                     skipVals,
                                     keepCacheObjects,
                                     deserializeBinary,
-                                    false);
+                                    false,
+                                    readVer);
                             }
                             else
                                 missed.put(key, txEntry.cached().version());
@@ -1600,7 +1638,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
 
     /**
      * Loads all missed keys for
-     * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean)}
method.
+     * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean,
boolean)} method.
      *
      * @param cacheCtx Cache context.
      * @param map Return map.
@@ -1721,7 +1759,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
         final boolean deserializeBinary,
         final boolean skipVals,
         final boolean keepCacheObjects,
-        final boolean skipStore) {
+        final boolean skipStore,
+        final boolean needVer) {
         if (F.isEmpty(keys))
             return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
 
@@ -1751,7 +1790,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
                 deserializeBinary,
                 skipVals,
                 keepCacheObjects,
-                skipStore);
+                skipStore,
+                needVer);
 
             if (single && missed.isEmpty())
                 return new GridFinishedFuture<>(retMap);
@@ -1797,25 +1837,48 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
implements Ig
                             while (true) {
                                 GridCacheEntryEx cached = txEntry.cached();
 
+                                CacheObject val = null;
+                                GridCacheVersion readVer = null;
+
                                 try {
                                     Object transformClo =
                                         (!F.isEmpty(txEntry.entryProcessors()) &&
                                             cctx.gridEvents().isRecordable(EVT_CACHE_OBJECT_READ))
?
                                             F.first(txEntry.entryProcessors()) : null;
 
-                                    CacheObject val = cached.innerGet(IgniteTxLocalAdapter.this,
-                                        cacheCtx.isSwapOrOffheapEnabled(),
-                                        /*read-through*/false,
-                                        /*fail-fast*/true,
-                                        /*unmarshal*/true,
-                                        /*metrics*/true,
-                                        /*events*/!skipVals,
-                                        /*temporary*/true,
-                                        CU.subjectId(IgniteTxLocalAdapter.this, cctx),
-                                        transformClo,
-                                        resolveTaskName(),
-                                        null,
-                                        txEntry.keepBinary());
+                                    if (needVer) {
+                                        T2<CacheObject, GridCacheVersion> res = cached.innerGetVersioned(
+                                            IgniteTxLocalAdapter.this,
+                                            /*swap*/cacheCtx.isSwapOrOffheapEnabled(),
+                                            /*unmarshal*/true,
+                                            /**update-metrics*/true,
+                                            /*event*/!skipVals,
+                                            CU.subjectId(IgniteTxLocalAdapter.this, cctx),
+                                            transformClo,
+                                            resolveTaskName(),
+                                            null,
+                                            txEntry.keepBinary());
+
+                                        if (res != null) {
+                                            val = res.get1();
+                                            readVer = res.get2();
+                                        }
+                                    }
+                                    else{
+                                        val = cached.innerGet(IgniteTxLocalAdapter.this,
+                                            cacheCtx.isSwapOrOffheapEnabled(),
+                                            /*read-through*/false,
+                                            /*fail-fast*/true,
+                                            /*unmarshal*/true,
+                                            /*metrics*/true,
+                                            /*events*/!skipVals,
+                                            /*temporary*/true,
+                                            CU.subjectId(IgniteTxLocalAdapter.this, cctx),
+                                            transformClo,
+                                            resolveTaskName(),
+                                            null,
+                                            txEntry.keepBinary());
+                                    }
 
                                     // If value is in cache and passed the filter.
                                     if (val != null) {
@@ -1832,7 +1895,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements
Ig
                                             skipVals,
                                             keepCacheObjects,
                                             deserializeBinary,
-                                            false);
+                                            false,
+                                            readVer);
                                     }
 
                                     // Even though we bring the value back from lock acquisition,

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
index a5d3373..78f517c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java
@@ -76,7 +76,8 @@ public interface IgniteTxLocalEx extends IgniteInternalTx {
         boolean deserializeBinary,
         boolean skipVals,
         boolean keepCacheObjects,
-        boolean skipStore);
+        boolean skipStore,
+        boolean needVer);
 
     /**
      * @param cacheCtx Cache context.

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntrySeltTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntrySeltTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntrySeltTest.java
new file mode 100644
index 0000000..a9131d9
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheGetEntrySeltTest.java
@@ -0,0 +1,206 @@
+/*
+ * 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.io.Serializable;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+
+/**
+ * Test getEntry and getEntries methods.
+ */
+public class CacheGetEntrySeltTest extends GridCacheAbstractSelfTest {
+
+    @Override protected int gridCount() {
+        return 3;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        CacheConfiguration cacheCfg1 = new CacheConfiguration();
+        cacheCfg1.setCacheMode(CacheMode.PARTITIONED);
+        cacheCfg1.setName("near");
+        cacheCfg1.setNearConfiguration(new NearCacheConfiguration());
+
+        CacheConfiguration cacheCfg1t = new CacheConfiguration();
+        cacheCfg1t.setCacheMode(CacheMode.PARTITIONED);
+        cacheCfg1t.setAtomicityMode(TRANSACTIONAL);
+        cacheCfg1t.setName("nearT");
+        cacheCfg1t.setNearConfiguration(new NearCacheConfiguration());
+
+        CacheConfiguration cacheCfg2 = new CacheConfiguration();
+        cacheCfg2.setCacheMode(CacheMode.PARTITIONED);
+        cacheCfg2.setName("partitioned");
+
+        CacheConfiguration cacheCfg2t = new CacheConfiguration();
+        cacheCfg2t.setCacheMode(CacheMode.PARTITIONED);
+        cacheCfg2t.setAtomicityMode(TRANSACTIONAL);
+        cacheCfg2t.setName("partitionedT");
+
+        CacheConfiguration cacheCfg3 = new CacheConfiguration();
+        cacheCfg3.setCacheMode(CacheMode.LOCAL);
+        cacheCfg3.setName("local");
+
+        CacheConfiguration cacheCfg3t = new CacheConfiguration();
+        cacheCfg3t.setCacheMode(CacheMode.LOCAL);
+        cacheCfg3t.setAtomicityMode(TRANSACTIONAL);
+        cacheCfg3t.setName("localT");
+
+        CacheConfiguration cacheCfg4 = new CacheConfiguration();
+        cacheCfg4.setCacheMode(CacheMode.REPLICATED);
+        cacheCfg4.setName("replicated");
+
+        CacheConfiguration cacheCfg4t = new CacheConfiguration();
+        cacheCfg4t.setCacheMode(CacheMode.REPLICATED);
+        cacheCfg4t.setAtomicityMode(TRANSACTIONAL);
+        cacheCfg4t.setName("replicatedT");
+
+        cfg.setMarshaller(null);
+
+        cfg.setCacheConfiguration(cfg.getCacheConfiguration()[0], cacheCfg1, cacheCfg1t,
cacheCfg2, cacheCfg2t,
+            cacheCfg3, cacheCfg3t, cacheCfg4, cacheCfg4t);
+
+        return cfg;
+    }
+
+    @Override protected long getTestTimeout() {
+        return Long.MAX_VALUE;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testGetEntry() throws Exception {
+        test0("near");
+        test0("nearT");
+        test0("partitioned");
+        test0("partitionedT");
+        test0("local");
+        test0("localT");
+        test0("replicated");
+        test0("replicatedT");
+    }
+
+    private void test0(String name) {
+        IgniteCache<Integer, TestValue> cache = grid(0).cache(name);
+
+        // Put.
+        for (int i = 0; i < 10_000; ++i)
+            cache.put(i, new TestValue(i));
+
+        // getEntry regular.
+        for (int i = 0; i < 10_000; ++i) {
+            CacheEntry<Integer, TestValue> e = cache.getEntry(i);
+
+            assertEquals(e.getValue().val, i);
+
+            assertNotNull(e.version());
+        }
+
+        // getEntries regular.
+        for (int i = 0; i < 10_000; ++i) {
+            Set<Integer> set = new HashSet<>();
+
+            for (int j = 0; j < 10; j++)
+                set.add(++i);
+
+            Collection<CacheEntry<Integer, TestValue>> es = cache.getEntries(set);
+
+            for (CacheEntry<Integer, TestValue> e : es) {
+                assertEquals((Integer)e.getValue().val, e.getKey());
+
+                assertTrue(e.getValue().val <= i);
+                assertTrue(e.getValue().val > i - 10);
+
+                assertNotNull(e.version());
+            }
+        }
+
+        IgniteCache<Integer, BinaryObject> cacheB = grid(0).cache(name).withKeepBinary();
+
+        // getEntry withKeepBinary.
+        for (int i = 0; i < 10_000; ++i) {
+            CacheEntry<Integer, BinaryObject> e = cacheB.getEntry(i);
+
+            assertEquals(((TestValue)e.getValue().deserialize()).val, i);
+
+            assertNotNull(e.version());
+        }
+
+        // getEntries withKeepBinary.
+        for (int i = 0; i < 10_000; ++i) {
+            Set<Integer> set = new HashSet<>();
+
+            for (int j = 0; j < 10; j++)
+                set.add(++i);
+
+            Collection<CacheEntry<Integer, BinaryObject>> es = cacheB.getEntries(set);
+
+            for (CacheEntry<Integer, BinaryObject> e : es) {
+                TestValue tv = e.getValue().deserialize();
+
+                assertEquals((Integer)tv.val, e.getKey());
+
+                assertTrue((tv).val <= i);
+                assertTrue((tv).val > i - 10);
+
+                assertNotNull(e.version());
+            }
+        }
+    }
+
+    /**
+     *
+     */
+    private static class TestValue implements Serializable {
+        /** */
+        private int val;
+
+        /**
+         * @param val Value.
+         */
+        public TestValue(int val) {
+            this.val = val;
+        }
+
+        /**
+         * @return Value.
+         */
+        public int value() {
+            return val;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(TestValue.class, this);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index 1e0071e..3f4f06c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -53,6 +53,7 @@ import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.IgniteTransactions;
+import org.apache.ignite.cache.CacheEntry;
 import org.apache.ignite.cache.CacheEntryProcessor;
 import org.apache.ignite.cache.CacheMemoryMode;
 import org.apache.ignite.cache.CachePeekMode;
@@ -556,6 +557,30 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
     /**
      * @throws Exception In case of error.
      */
+    public void testGetEntry() throws Exception {
+        IgniteCache<String, Integer> cache = jcache();
+
+        cache.put("key1", 1);
+        cache.put("key2", 2);
+
+        CacheEntry<String, Integer> key1e =  cache.getEntry("key1");
+        CacheEntry<String, Integer> key2e =  cache.getEntry("key2");
+        CacheEntry<String, Integer> wrongKeye =  cache.getEntry("wrongKey");
+
+        assert key1e.getValue() == 1;
+        assert key1e.getKey().equals("key1");
+        assert key1e.version() != null;
+
+        assert key2e.getValue() == 2;
+        assert key2e.getKey().equals("key2");
+        assert key2e.version() != null;
+
+        assert wrongKeye == null;
+    }
+
+    /**
+     * @throws Exception In case of error.
+     */
     public void testGetAsync() throws Exception {
         IgniteCache<String, Integer> cache = jcache();
 
@@ -664,6 +689,122 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
     /**
      * @throws Exception In case of error.
      */
+    public void testGetEntries() throws Exception {
+        Transaction tx = txShouldBeUsed() ? transactions().txStart() : null;
+
+        final IgniteCache<String, Integer> cache = jcache();
+
+        try {
+            cache.put("key1", 1);
+            cache.put("key2", 2);
+
+            if (tx != null)
+                tx.commit();
+        }
+        finally {
+            if (tx != null)
+                tx.close();
+        }
+
+        GridTestUtils.assertThrows(log, new Callable<Void>() {
+            @Override public Void call() throws Exception {
+                cache.getAll(null).isEmpty();
+
+                return null;
+            }
+        }, NullPointerException.class, null);
+
+        assert cache.getEntries(Collections.<String>emptySet()).isEmpty();
+
+        Collection<CacheEntry<String, Integer>> c1 = cache.getEntries(ImmutableSet.of("key1",
"key2", "key9999"));
+
+        info("Retrieved c1: " + c1);
+
+        assert 2 == c1.size() : "Invalid collection: " + c1;
+
+        boolean b1 = false;
+        boolean b2 = false;
+
+        for (CacheEntry<String, Integer> e: c1){
+            if (e.getKey().equals("key1") && e.getValue().equals(1))
+                b1 = true;
+
+            if (e.getKey().equals("key2") && e.getValue().equals(2))
+                b2 = true;
+        }
+
+        assertTrue(b1 && b2);
+
+        Collection<CacheEntry<String, Integer>> c2 = cache.getEntries(ImmutableSet.of("key1",
"key2", "key9999"));
+
+        info("Retrieved c2: " + c2);
+
+        assert 2 == c2.size() : "Invalid collection: " + c2;
+
+        b1 = false;
+        b2 = false;
+
+        for (CacheEntry<String, Integer> e: c2){
+            if (e.getKey().equals("key1") && e.getValue().equals(1))
+                b1 = true;
+
+            if (e.getKey().equals("key2") && e.getValue().equals(2))
+                b2 = true;
+        }
+
+        assertTrue(b1 && b2);
+
+        // Now do the same checks but within transaction.
+        if (txShouldBeUsed()) {
+            try (Transaction tx0 = transactions().txStart()) {
+                assert cache.getEntries(Collections.<String>emptySet()).isEmpty();
+
+                c1 = cache.getEntries(ImmutableSet.of("key1", "key2", "key9999"));
+
+                info("Retrieved c1: " + c1);
+
+                assert 2 == c1.size() : "Invalid collection: " + c1;
+
+                b1 = false;
+                b2 = false;
+
+                for (CacheEntry<String, Integer> e: c1){
+                    if (e.getKey().equals("key1") && e.getValue().equals(1))
+                        b1 = true;
+
+                    if (e.getKey().equals("key2") && e.getValue().equals(2))
+                        b2 = true;
+                }
+
+                assertTrue(b1 && b2);
+
+                c2 = cache.getEntries(ImmutableSet.of("key1", "key2", "key9999"));
+
+                info("Retrieved c2: " + c2);
+
+                assert 2 == c2.size() : "Invalid collection: " + c2;
+
+                b1 = false;
+                b2 = false;
+
+                for (CacheEntry<String, Integer> e: c2){
+                    if (e.getKey().equals("key1") && e.getValue().equals(1))
+                        b1 = true;
+
+                    if (e.getKey().equals("key2") && e.getValue().equals(2))
+                        b2 = true;
+                }
+
+                assertTrue(b1 && b2);
+
+                tx0.commit();
+            }
+        }
+    }
+
+    /**
+     * @throws Exception In case of error.
+     */
     public void testGetAllWithNulls() throws Exception {
         final IgniteCache<String, Integer> cache = jcache();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
index c57869f..99203b1 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.cache;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -29,6 +30,7 @@ import javax.cache.processor.EntryProcessor;
 import javax.cache.processor.MutableEntry;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.cache.CacheAtomicWriteOrderMode;
+import org.apache.ignite.cache.CacheEntry;
 import org.apache.ignite.cache.CacheInterceptor;
 import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cache.affinity.Affinity;
@@ -257,6 +259,112 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
     /**
      * @throws Exception If failed.
      */
+    public void testGetEntry() throws Exception {
+        testGetEntry(primaryKey(0));
+
+        afterTest();
+
+        if (cacheMode() != LOCAL)
+            testGetEntry(backupKey(0));
+    }
+
+    /**
+     * @param key Key.
+     * @throws Exception If failed.
+     */
+    private void testGetEntry(String key) throws Exception {
+        // Try when value is not in cache.
+
+        interceptor.retInterceptor = new NullGetInterceptor();
+
+        log.info("Get 1.");
+
+        IgniteCache<String, Integer> cache = jcache(0);
+
+        assertEquals(null, cache.getEntry(key).getValue());
+
+        assertEquals(1, interceptor.invokeCnt.get());
+
+        assertEquals(0, interceptor.getMap.size());
+
+        interceptor.reset();
+
+        interceptor.retInterceptor = new OneGetInterceptor();
+
+        log.info("Get 2.");
+
+        assertEquals((Integer)1, cache.getEntry(key).getValue());
+
+        assertEquals(1, interceptor.invokeCnt.get());
+
+        assertEquals(0, interceptor.getMap.size());
+
+        interceptor.reset();
+
+        // Disable interceptor and update cache.
+
+        interceptor.disabled = true;
+
+        cache.put(key, 100);
+
+        interceptor.disabled = false;
+
+        // Try when value is in cache.
+
+        interceptor.retInterceptor = new NullGetInterceptor();
+
+        log.info("Get 3.");
+
+        assertEquals(null, cache.getEntry(key).getValue());
+
+        assertEquals(1, interceptor.invokeCnt.get());
+
+        assertEquals(1, interceptor.getMap.size());
+
+        assertEquals(100, interceptor.getMap.get(key));
+
+        checkCacheValue(key, 100);
+
+        interceptor.reset();
+
+        interceptor.retInterceptor = new GetIncrementInterceptor();
+
+        log.info("Get 4.");
+
+        assertEquals((Integer)101, cache.getEntry(key).getValue());
+
+        assertEquals(1, interceptor.invokeCnt.get());
+
+        assertEquals(1, interceptor.getMap.size());
+
+        assertEquals(100, interceptor.getMap.get(key));
+
+        checkCacheValue(key, 100);
+
+        interceptor.reset();
+
+        interceptor.retInterceptor = new GetIncrementInterceptor();
+
+        log.info("GetAsync 1.");
+
+        IgniteCache<String, Integer> cacheAsync = cache.withAsync();
+
+        cacheAsync.getEntry(key);
+
+        assertEquals((Integer)101, cacheAsync.<CacheEntry<String, Integer>>future().get().getValue());
+
+        assertEquals(1, interceptor.invokeCnt.get());
+
+        assertEquals(1, interceptor.getMap.size());
+
+        assertEquals(100, interceptor.getMap.get(key));
+
+        checkCacheValue(key, 100);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
     public void testGetAll() throws Exception {
         Set<String> keys = new LinkedHashSet<>();
 
@@ -350,6 +458,90 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
     /**
      * @throws Exception If failed.
      */
+    public void testGetEntries() throws Exception {
+        Set<String> keys = new LinkedHashSet<>();
+
+        for (int i = 0; i < 1000; i++)
+            keys.add(String.valueOf(i));
+
+        interceptor.retInterceptor = new NullGetInterceptor();
+
+        IgniteCache<String, Integer> cache = jcache(0);
+
+        IgniteCache<String, Integer> cacheAsync = cache.withAsync();
+
+        Collection<CacheEntry<String, Integer>> c = cache.getEntries(keys);
+
+        assertEquals(0, c.size());
+
+        assertEquals(1000, interceptor.invokeCnt.get());
+
+        interceptor.reset();
+
+        interceptor.retInterceptor = new GetAllInterceptor1();
+
+        c = cache.getEntries(keys);
+
+        assertEquals(500, c.size());
+
+        for (CacheEntry<String, Integer> e : c) {
+            int k = Integer.valueOf(e.getKey());
+
+            assertEquals((Integer)(k * 2), e.getValue());
+        }
+
+        assertEquals(1000, interceptor.invokeCnt.get());
+
+        // Put some values in cache.
+
+        interceptor.disabled = true;
+
+        for (int i = 0; i < 500; i++)
+            cache.put(String.valueOf(i), i);
+
+        interceptor.disabled = false;
+
+        for (int j = 0; j < 2; j++) {
+            interceptor.reset();
+
+            interceptor.retInterceptor = new GetAllInterceptor2();
+
+            if (j == 0)
+                c = cache.getEntries(keys);
+            else {
+                cacheAsync.getEntries(keys);
+
+                c = cacheAsync.<Collection<CacheEntry<String, Integer>>>future().get();
+            }
+
+            for (CacheEntry<String, Integer> e : c) {
+                int k = Integer.valueOf(e.getKey());
+
+                switch (k % 3) {
+                    case 1:
+                        Integer exp = k < 500 ? k : null;
+
+                        assertEquals(exp, e.getValue());
+
+                        break;
+
+                    case 2:
+                        assertEquals((Integer)(k * 3), e.getValue());
+
+                        break;
+
+                    default:
+                        fail();
+                }
+            }
+
+            assertEquals(1000, interceptor.invokeCnt.get());
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
     public void testCancelUpdate() throws Exception {
         for (Operation op : Operation.values()) {
             testCancelUpdate(primaryKey(0), op);

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
index 6a7416e..293ba1e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtEvictionNearReadersSelfTest.java
@@ -263,7 +263,7 @@ public class GridCacheDhtEvictionNearReadersSelfTest extends GridCommonAbstractT
             waitForLocalEvent(grid(primary).events(), nodeEvent(primary.id()), EVT_CACHE_ENTRY_EVICTED);
 
         // Get value on other node, it should be loaded to near cache.
-        assertEquals(val, nearOther.get(key, true));
+        assertEquals(val, nearOther.get(key, true, false));
 
         entryPrimary = (GridDhtCacheEntry)dhtPrimary.peekEx(key);
         entryBackup = (GridDhtCacheEntry)dhtBackup.peekEx(key);

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
index 035f1b0..f6d0c41 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteCacheProcessProxy.java
@@ -34,6 +34,7 @@ import javax.cache.processor.EntryProcessorResult;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.cache.CacheEntry;
 import org.apache.ignite.cache.CacheEntryProcessor;
 import org.apache.ignite.cache.CacheMetrics;
 import org.apache.ignite.cache.CachePeekMode;
@@ -129,7 +130,7 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K,
V> {
     }
 
     /** {@inheritDoc} */
-    @Override public void loadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable
Object... args) 
+    @Override public void loadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable
Object... args)
         throws CacheException {
         throw new UnsupportedOperationException("Method should be supported.");
     }
@@ -224,11 +225,21 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K,
V> {
     }
 
     /** {@inheritDoc} */
+    @Override public CacheEntry<K, V> getEntry(K key) {
+        return compute.call(new GetEntryTask<K, V>(cacheName, isAsync, key));
+    }
+
+    /** {@inheritDoc} */
     @Override public Map<K, V> getAll(Set<? extends K> keys) {
         return compute.call(new GetAllTask<K, V>(cacheName, isAsync, keys));
     }
 
     /** {@inheritDoc} */
+    @Override public Collection<CacheEntry<K, V>> getEntries(Set<? extends
K> keys) {
+        return compute.call(new GetEntriesTask<K, V>(cacheName, isAsync, keys));
+    }
+
+    /** {@inheritDoc} */
     @Override public Map<K, V> getAllOutTx(Set<? extends K> keys) {
         return compute.call(new GetAllOutTxTask<K, V>(cacheName, isAsync, keys));
     }
@@ -710,6 +721,30 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K,
V> {
     /**
      *
      */
+    private static class GetEntryTask<K, V> extends CacheTaskAdapter<K, V, CacheEntry<K,
V>> {
+        /** Key. */
+        private final K key;
+
+        /**
+         * @param cacheName Cache name.
+         * @param async Async.
+         * @param key Key.
+         */
+        public GetEntryTask(String cacheName, boolean async, K key) {
+            super(cacheName, async);
+            this.key = key;
+        }
+
+        /** {@inheritDoc} */
+        @Override public CacheEntry<K, V> call() throws Exception {
+            return cache().getEntry(key);
+        }
+    }
+
+
+    /**
+     *
+     */
     private static class RemoveAllTask<K, V> extends CacheTaskAdapter<K, V, Void>
{
         /**
          * @param cacheName Cache name.
@@ -973,6 +1008,29 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K,
V> {
     /**
      *
      */
+    private static class GetEntriesTask<K, V> extends CacheTaskAdapter<K, V, Collection<CacheEntry<K,
V>> > {
+        /** Keys. */
+        private final Set<? extends K> keys;
+
+        /**
+         * @param cacheName Cache name.
+         * @param async Async.
+         * @param keys Keys.
+         */
+        public GetEntriesTask(String cacheName, boolean async, Set<? extends K> keys)
{
+            super(cacheName, async);
+            this.keys = keys;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Collection<CacheEntry<K, V>>  call() throws Exception
{
+            return cache().getEntries(keys);
+        }
+    }
+
+    /**
+     *
+     */
     private static class GetAllOutTxTask<K, V> extends CacheTaskAdapter<K, V, Map<K,
V>> {
         /** Keys. */
         private final Set<? extends K> keys;

http://git-wip-us.apache.org/repos/asf/ignite/blob/920089ce/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index 68e52df..8fb5e14 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -19,6 +19,7 @@ package org.apache.ignite.testsuites;
 
 import junit.framework.TestSuite;
 import org.apache.ignite.cache.store.jdbc.CacheJdbcStoreSessionListenerSelfTest;
+import org.apache.ignite.internal.processors.cache.CacheGetEntrySeltTest;
 import org.apache.ignite.internal.processors.cache.IgniteCacheGetCustomCollectionsSelfTest;
 import org.apache.ignite.internal.processors.GridCacheTxLoadFromStoreOnLockSelfTest;
 import org.apache.ignite.internal.processors.cache.CacheClientStoreSelfTest;
@@ -241,6 +242,7 @@ public class IgniteCacheTestSuite4 extends TestSuite {
         suite.addTestSuite(IgniteSystemCacheOnClientTest.class);
 
         suite.addTestSuite(CacheRemoveAllSelfTest.class);
+        suite.addTestSuite(CacheGetEntrySeltTest.class);
 
         suite.addTestSuite(CacheStopAndDestroySelfTest.class);
 


Mime
View raw message