ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From akuznet...@apache.org
Subject [23/50] [abbrv] incubator-ignite git commit: # ignite-51 merge from sprint-2
Date Wed, 11 Mar 2015 11:07:55 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index c2a00e6,0435b92..94bac05
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@@ -81,11 -80,8 +80,8 @@@ public final class GridNearGetFuture<K
      private GridCacheVersion ver;
  
      /** Transaction. */
 -    private IgniteTxLocalEx<K, V> tx;
 +    private IgniteTxLocalEx tx;
  
-     /** Logger. */
-     private IgniteLogger log;
- 
      /** Trackable flag. */
      private boolean trackable;
  
@@@ -490,13 -494,14 +480,13 @@@
                  }
  
                  if (v != null && !reload) {
 -                    K key0 = key;
 +                    K key0 = key.value(cctx.cacheObjectContext(), true);
 +                    V val0 = v.value(cctx.cacheObjectContext(), true);
  
 -                    if (cctx.portableEnabled()) {
 -                        v = (V)cctx.unwrapPortableIfNeeded(v, !deserializePortable);
 -                        key0 = (K)cctx.unwrapPortableIfNeeded(key, !deserializePortable);
 -                    }
 +                    val0 = (V)cctx.unwrapPortableIfNeeded(val0, !deserializePortable);
 +                    key0 = (K)cctx.unwrapPortableIfNeeded(key0, !deserializePortable);
  
-                     add(new GridFinishedFuture<>(cctx.kernalContext(), Collections.singletonMap(key0, val0)));
 -                    add(new GridFinishedFuture<>(Collections.singletonMap(key0, v)));
++                    add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
                  }
                  else {
                      if (primary == null)
@@@ -682,12 -683,7 +662,10 @@@
           * @param savedVers Saved entry versions.
           * @param topVer Topology version.
           */
 -        MiniFuture(ClusterNode node, LinkedHashMap<K, Boolean> keys, Map<K, GridCacheVersion> savedVers, long topVer) {
 +        MiniFuture(ClusterNode node,
 +            LinkedHashMap<KeyCacheObject, Boolean> keys,
 +            Map<KeyCacheObject, GridCacheVersion> savedVers,
 +            long topVer) {
-             super(cctx.kernalContext());
- 
              this.node = node;
              this.keys = keys;
              this.savedVers = savedVers;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
index 0ffc465,88095d7..af50959
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
@@@ -49,10 -48,7 +48,7 @@@ import static org.apache.ignite.events.
   * Cache lock future.
   */
  public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<Boolean>
 -    implements GridCacheMvccFuture<K, V, Boolean> {
 +    implements GridCacheMvccFuture<Boolean> {
-     /** */
-     private static final long serialVersionUID = 0L;
- 
      /** Logger reference. */
      private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
  
@@@ -92,12 -91,8 +91,8 @@@
      /** Lock timeout. */
      private long timeout;
  
-     /** Logger. */
-     @GridToStringExclude
-     private IgniteLogger log;
- 
      /** Filter. */
 -    private IgnitePredicate<Cache.Entry<K, V>>[] filter;
 +    private CacheEntryPredicate[] filter;
  
      /** Transaction. */
      @GridToStringExclude
@@@ -961,14 -950,12 +950,12 @@@
              if (log.isDebugEnabled())
                  log.debug("Before locally locking near request: " + req);
  
 -            IgniteInternalFuture<GridNearLockResponse<K, V>> fut = dht().lockAllAsync(cctx, cctx.localNode(), req, filter);
 +            IgniteInternalFuture<GridNearLockResponse> fut = dht().lockAllAsync(cctx, cctx.localNode(), req, filter);
  
              // Add new future.
-             add(new GridEmbeddedFuture(
-                 cctx.kernalContext(),
-                 fut,
+             add(new GridEmbeddedFuture<>(
 -                new C2<GridNearLockResponse<K, V>, Exception, Boolean>() {
 -                    @Override public Boolean apply(GridNearLockResponse<K, V> res, Exception e) {
 +                new C2<GridNearLockResponse, Exception, Boolean>() {
 +                    @Override public Boolean apply(GridNearLockResponse res, Exception e) {
                          if (CU.isLockTimeoutOrCancelled(e) ||
                              (res != null && CU.isLockTimeoutOrCancelled(res.error())))
                              return false;
@@@ -1254,10 -1234,8 +1231,8 @@@
           * @param keys Keys.
           * @param mappings Mappings to proceed.
           */
 -        MiniFuture(ClusterNode node, Collection<K> keys,
 -            ConcurrentLinkedDeque8<GridNearLockMapping<K, V>> mappings) {
 +        MiniFuture(ClusterNode node, Collection<KeyCacheObject> keys,
 +            ConcurrentLinkedDeque8<GridNearLockMapping> mappings) {
-             super(cctx.kernalContext());
- 
              this.node = node;
              this.keys = keys;
              this.mappings = mappings;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
index ee80a00,ff595c0..c718a67
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
@@@ -109,12 -109,9 +109,12 @@@ public class GridNearTransactionalCache
          ctx.checkSecurity(GridSecurityPermission.CACHE_READ);
  
          if (F.isEmpty(keys))
-             return new GridFinishedFuture<>(ctx.kernalContext(), Collections.<K, V>emptyMap());
+             return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
  
 -        IgniteTxLocalAdapter<K, V> tx = ctx.tm().threadLocalTx();
 +        if (keyCheck)
 +            validateCacheKeys(keys);
 +
 +        IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx();
  
          if (tx != null && !tx.implicit() && !skipTx) {
              return asyncOp(tx, new AsyncOp<Map<K, V>>(keys) {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index a8804dc,dd5d07d..d338352
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@@ -421,21 -408,12 +408,12 @@@ public final class GridNearTxFinishFutu
  
          /** Keys. */
          @GridToStringInclude
 -        private GridDistributedTxMapping<K, V> m;
 +        private GridDistributedTxMapping m;
  
          /**
-          * Empty constructor required for {@link Externalizable}.
-          */
-         public MiniFuture() {
-             // No-op.
-         }
- 
-         /**
           * @param m Mapping.
           */
 -        MiniFuture(GridDistributedTxMapping<K, V> m) {
 +        MiniFuture(GridDistributedTxMapping m) {
-             super(cctx.kernalContext());
- 
              this.m = m;
          }
  

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index 6669773,9ada718..4477587
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@@ -761,11 -759,11 +760,11 @@@ public class GridNearTxLocal extends Gr
  
          cctx.mvcc().addFuture(fut);
  
 -        IgniteInternalFuture<IgniteInternalTx<K, V>> prepareFut = prepFut.get();
 +        IgniteInternalFuture<IgniteInternalTx> prepareFut = prepFut.get();
  
-         prepareFut.listenAsync(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 -        prepareFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx<K, V>>>() {
 -            @Override public void apply(IgniteInternalFuture<IgniteInternalTx<K, V>> f) {
 -                GridNearTxFinishFuture<K, V> fut0 = commitFut.get();
++        prepareFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 +            @Override public void apply(IgniteInternalFuture<IgniteInternalTx> f) {
 +                GridNearTxFinishFuture fut0 = commitFut.get();
  
                  try {
                      // Make sure that here are no exceptions.
@@@ -832,8 -830,8 +831,8 @@@
              }
          }
          else {
-             prepFut.listenAsync(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 -            prepFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx<K, V>>>() {
 -                @Override public void apply(IgniteInternalFuture<IgniteInternalTx<K, V>> f) {
++            prepFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 +                @Override public void apply(IgniteInternalFuture<IgniteInternalTx> f) {
                      try {
                          // Check for errors in prepare future.
                          f.get();
@@@ -966,10 -964,10 +965,10 @@@
              if (prep != null)
                  return (IgniteInternalFuture<IgniteInternalTx>)(IgniteInternalFuture)prep;
  
-             return new GridFinishedFuture<IgniteInternalTx>(cctx.kernalContext(), this);
+             return new GridFinishedFuture<IgniteInternalTx>(this);
          }
  
 -        final GridDhtTxFinishFuture<K, V> fut = new GridDhtTxFinishFuture<>(cctx, this, /*commit*/true);
 +        final GridDhtTxFinishFuture fut = new GridDhtTxFinishFuture<>(cctx, this, /*commit*/true);
  
          cctx.mvcc().addFuture(fut);
  
@@@ -995,8 -993,8 +994,8 @@@
              }
          }
          else
-             prep.listenAsync(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 -            prep.listen(new CI1<IgniteInternalFuture<IgniteInternalTx<K, V>>>() {
 -                @Override public void apply(IgniteInternalFuture<IgniteInternalTx<K, V>> f) {
++            prep.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 +                @Override public void apply(IgniteInternalFuture<IgniteInternalTx> f) {
                      try {
                          f.get(); // Check for errors of a parent future.
  
@@@ -1048,8 -1046,8 +1047,8 @@@
              fut.finish();
          }
          else
-             prep.listenAsync(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 -            prep.listen(new CI1<IgniteInternalFuture<IgniteInternalTx<K, V>>>() {
 -                @Override public void apply(IgniteInternalFuture<IgniteInternalTx<K, V>> f) {
++            prep.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 +                @Override public void apply(IgniteInternalFuture<IgniteInternalTx> f) {
                      try {
                          f.get(); // Check for errors of a parent future.
                      }
@@@ -1077,13 -1075,13 +1076,13 @@@
              checkValid();
          }
          catch (IgniteCheckedException e) {
-             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+             return new GridFinishedFuture<>(e);
          }
  
 -        final GridCacheReturn<V> ret = new GridCacheReturn<>(false);
 +        final GridCacheReturn ret = new GridCacheReturn(localResult(), false);
  
          if (F.isEmpty(keys))
-             return new GridFinishedFuture<>(cctx.kernalContext(), ret);
+             return new GridFinishedFuture<>(ret);
  
          init();
  

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
index 3a86888,cc5fcf4..778570a
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
@@@ -50,11 -50,8 +49,8 @@@ import static org.apache.ignite.transac
  /**
   *
   */
 -public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFuture<IgniteInternalTx<K, V>>
 -    implements GridCacheMvccFuture<K, V, IgniteInternalTx<K, V>> {
 +public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFuture<IgniteInternalTx>
 +    implements GridCacheMvccFuture<IgniteInternalTx> {
-     /** */
-     private static final long serialVersionUID = 0L;
- 
      /** Logger reference. */
      private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
  
@@@ -66,11 -66,8 +65,8 @@@
  
      /** Transaction. */
      @GridToStringExclude
 -    private GridNearTxLocal<K, V> tx;
 +    private GridNearTxLocal tx;
  
-     /** Logger. */
-     private IgniteLogger log;
- 
      /** Error. */
      @GridToStringExclude
      private AtomicReference<Throwable> err = new AtomicReference<>(null);
@@@ -82,16 -79,9 +78,9 @@@
      private GridDhtTxMapping<K, V> txMapping;
  
      /** */
 -    private Collection<IgniteTxKey<K>> lockKeys = new GridConcurrentHashSet<>();
 +    private Collection<IgniteTxKey> lockKeys = new GridConcurrentHashSet<>();
  
      /**
-      * Empty constructor required for {@link Externalizable}.
-      */
-     public GridNearTxPrepareFuture() {
-         // No-op.
-     }
- 
-     /**
       * @param cctx Context.
       * @param tx Transaction.
       */
@@@ -828,11 -821,8 +820,8 @@@
       * Mini-future for get operations. Mini-futures are only waiting on a single
       * node as opposed to multiple nodes.
       */
 -    private class MiniFuture extends GridFutureAdapter<IgniteInternalTx<K, V>> {
 +    private class MiniFuture extends GridFutureAdapter<IgniteInternalTx> {
          /** */
-         private static final long serialVersionUID = 0L;
- 
-         /** */
          private final IgniteUuid futId = IgniteUuid.randomUuid();
  
          /** Keys. */
@@@ -843,23 -833,16 +832,16 @@@
          private AtomicBoolean rcvRes = new AtomicBoolean(false);
  
          /** Mappings to proceed prepare. */
 -        private ConcurrentLinkedDeque8<GridDistributedTxMapping<K, V>> mappings;
 +        private ConcurrentLinkedDeque8<GridDistributedTxMapping> mappings;
  
          /**
-          * Empty constructor required for {@link Externalizable}.
-          */
-         public MiniFuture() {
-             // No-op.
-         }
- 
-         /**
           * @param m Mapping.
           * @param mappings Queue of mappings to proceed with.
           */
-         MiniFuture(GridDistributedTxMapping m,
-             ConcurrentLinkedDeque8<GridDistributedTxMapping> mappings) {
-             super(cctx.kernalContext());
- 
+         MiniFuture(
 -            GridDistributedTxMapping<K, V> m,
 -            ConcurrentLinkedDeque8<GridDistributedTxMapping<K, V>> mappings
++            GridDistributedTxMapping m,
++            ConcurrentLinkedDeque8<GridDistributedTxMapping> mappings
+         ) {
              this.m = m;
              this.mappings = mappings;
          }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
index 5eb4284,5833270..f493dbc
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
@@@ -131,12 -124,10 +131,12 @@@ public class GridLocalCache<K, V> exten
       * @param filter Filter.
       * @return Future.
       */
 -    public IgniteInternalFuture<Boolean> lockAllAsync(Collection<? extends K> keys, long timeout,
 -        @Nullable IgniteTxLocalEx<K, V> tx, IgnitePredicate<Cache.Entry<K, V>>[] filter) {
 +    public IgniteInternalFuture<Boolean> lockAllAsync(Collection<KeyCacheObject> keys,
 +        long timeout,
 +        @Nullable IgniteTxLocalEx tx,
 +        CacheEntryPredicate[] filter) {
          if (F.isEmpty(keys))
-             return new GridFinishedFuture<>(ctx.kernalContext(), true);
+             return new GridFinishedFuture<>(true);
  
          GridLocalLockFuture<K, V> fut = new GridLocalLockFuture<>(ctx, keys, tx, this, timeout, filter);
  

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
index 8a8fb5c,320a443..dad40fd
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
@@@ -38,10 -37,7 +37,7 @@@ import java.util.concurrent.atomic.*
   * Cache lock future.
   */
  public final class GridLocalLockFuture<K, V> extends GridFutureAdapter<Boolean>
 -    implements GridCacheMvccFuture<K, V, Boolean> {
 +    implements GridCacheMvccFuture<Boolean> {
-     /** */
-     private static final long serialVersionUID = 0L;
- 
      /** Logger reference. */
      private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
  
@@@ -77,15 -76,11 +76,11 @@@
      /** Lock timeout. */
      private long timeout;
  
-     /** Logger. */
-     @GridToStringExclude
-     private IgniteLogger log;
- 
      /** Filter. */
 -    private IgnitePredicate<Cache.Entry<K, V>>[] filter;
 +    private CacheEntryPredicate[] filter;
  
      /** Transaction. */
 -    private IgniteTxLocalEx<K, V> tx;
 +    private IgniteTxLocalEx tx;
  
      /** Trackable flag. */
      private boolean trackable = true;
@@@ -107,13 -95,11 +95,11 @@@
       */
      GridLocalLockFuture(
          GridCacheContext<K, V> cctx,
 -        Collection<? extends K> keys,
 -        IgniteTxLocalEx<K, V> tx,
 +        Collection<KeyCacheObject> keys,
 +        IgniteTxLocalEx tx,
          GridLocalCache<K, V> cache,
          long timeout,
 -        IgnitePredicate<Cache.Entry<K, V>>[] filter) {
 +        CacheEntryPredicate[] filter) {
-         super(cctx.kernalContext());
- 
          assert keys != null;
          assert cache != null;
  

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTx.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTx.java
index 35b0a12,da7f73b..33e6174
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTx.java
@@@ -113,10 -111,10 +113,10 @@@ class GridLocalTx extends IgniteTxLocal
          try {
              prepare();
  
-             return new GridFinishedFuture<IgniteInternalTx>(cctx.kernalContext(), this);
 -            return new GridFinishedFuture<IgniteInternalTx<K, V>>(this);
++            return new GridFinishedFuture<IgniteInternalTx>(this);
          }
          catch (IgniteCheckedException e) {
-             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+             return new GridFinishedFuture<>(e);
          }
      }
  
@@@ -155,10 -153,10 +155,10 @@@
          catch (IgniteCheckedException e) {
              state(UNKNOWN);
  
-             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+             return new GridFinishedFuture<>(e);
          }
  
 -        GridLocalTxFuture<K, V> fut = this.fut.get();
 +        GridLocalTxFuture fut = this.fut.get();
  
          if (fut == null) {
              if (this.fut.compareAndSet(null, fut = new GridLocalTxFuture<>(cctx, this))) {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTxFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTxFuture.java
index bc248aa,c80292c..daf22c6
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTxFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalTxFuture.java
@@@ -37,11 -36,8 +36,8 @@@ import static org.apache.ignite.transac
  /**
   * Replicated cache transaction future.
   */
 -final class GridLocalTxFuture<K, V> extends GridFutureAdapter<IgniteInternalTx<K, V>>
 -    implements GridCacheMvccFuture<K, V, IgniteInternalTx<K, V>> {
 +final class GridLocalTxFuture<K, V> extends GridFutureAdapter<IgniteInternalTx>
 +    implements GridCacheMvccFuture<IgniteInternalTx> {
-     /** */
-     private static final long serialVersionUID = 0L;
- 
      /** Logger reference. */
      private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
  
@@@ -82,9 -70,7 +70,7 @@@
       */
      GridLocalTxFuture(
          GridCacheSharedContext<K, V> cctx,
 -        GridLocalTx<K, V> tx) {
 +        GridLocalTx tx) {
-         super(cctx.kernalContext());
- 
          assert cctx != null;
          assert tx != null;
  

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 8fb8ab4,6e0a85f..248515c
--- 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
@@@ -1532,8 -1553,8 +1532,8 @@@ public class GridLocalAtomicCache<K, V
          TransactionIsolation isolation,
          boolean invalidate,
          long accessTtl,
 -        IgnitePredicate<Cache.Entry<K, V>>[] filter) {
 +        CacheEntryPredicate[] filter) {
-         return new GridFinishedFutureEx<>(new UnsupportedOperationException("Locks are not supported for " +
+         return new GridFinishedFuture<>(new UnsupportedOperationException("Locks are not supported for " +
              "CacheAtomicityMode.ATOMIC mode (use CacheAtomicityMode.TRANSACTIONAL instead)"));
      }
  
@@@ -1541,8 -1562,8 +1541,8 @@@
      @SuppressWarnings("unchecked")
      @Override public IgniteInternalFuture<Boolean> lockAllAsync(@Nullable Collection<? extends K> keys,
          long timeout,
 -        @Nullable IgnitePredicate<Cache.Entry<K, V>>... filter) {
 +        @Nullable CacheEntryPredicate... filter) {
-         return new GridFinishedFutureEx<>(new UnsupportedOperationException("Locks are not supported for " +
+         return new GridFinishedFuture<>(new UnsupportedOperationException("Locks are not supported for " +
              "CacheAtomicityMode.ATOMIC mode (use CacheAtomicityMode.TRANSACTIONAL instead)"));
      }
  

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheLocalQueryFuture.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 2b7ecb6,b69c6d9..ecdc939
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@@ -167,19 -167,18 +167,18 @@@ public class IgniteTxHandler 
       * @param req Near prepare request.
       * @return Prepare future.
       */
 -    private IgniteInternalFuture<IgniteInternalTx<K, V>> prepareColocatedTx(
 -        final GridNearTxLocal<K, V> locTx,
 -        final GridNearTxPrepareRequest<K, V> req,
 -        final IgniteInClosure<GridNearTxPrepareResponse<K, V>> completeCb
 +    private IgniteInternalFuture<IgniteInternalTx> prepareColocatedTx(
 +        final GridNearTxLocal locTx,
 +        final GridNearTxPrepareRequest req,
 +        final IgniteInClosure<GridNearTxPrepareResponse> completeCb
      ) {
  
-         IgniteInternalFuture<Object> fut = new GridFinishedFutureEx<>(); // TODO force preload keys.
+         IgniteInternalFuture<Object> fut = new GridFinishedFuture<>(); // TODO force preload keys.
  
          return new GridEmbeddedFuture<>(
-             ctx.kernalContext(),
              fut,
 -            new C2<Object, Exception, IgniteInternalFuture<IgniteInternalTx<K, V>>>() {
 -                @Override public IgniteInternalFuture<IgniteInternalTx<K, V>> apply(Object o, Exception ex) {
 +            new C2<Object, Exception, IgniteInternalFuture<IgniteInternalTx>>() {
 +                @Override public IgniteInternalFuture<IgniteInternalTx> apply(Object o, Exception ex) {
                      if (ex != null)
                          throw new GridClosureException(ex);
  
@@@ -240,10 -239,10 +239,10 @@@
                  e.unmarshal(ctx, false, ctx.deploy().globalLoader());
          }
          catch (IgniteCheckedException e) {
-             return new GridFinishedFuture<>(ctx.kernalContext(), e);
+             return new GridFinishedFuture<>(e);
          }
  
 -        GridDhtTxLocal<K, V> tx;
 +        GridDhtTxLocal tx;
  
          GridCacheVersion mappedVer = ctx.tm().mappedVersion(req.version());
  
@@@ -324,10 -323,10 +323,10 @@@
                  }
              }
  
 -            final GridDhtTxLocal<K, V> tx0 = tx;
 +            final GridDhtTxLocal tx0 = tx;
  
-             fut.listenAsync(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 -            fut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx<K, V>>>() {
 -                @Override public void apply(IgniteInternalFuture<IgniteInternalTx<K, V>> txFut) {
++            fut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
 +                @Override public void apply(IgniteInternalFuture<IgniteInternalTx> txFut) {
                      try {
                          txFut.get();
                      }
@@@ -343,7 -342,7 +342,7 @@@
              return fut;
          }
          else
-             return new GridFinishedFuture<>(ctx.kernalContext(), (IgniteInternalTx)null);
 -            return new GridFinishedFuture<>((IgniteInternalTx<K, V>)null);
++            return new GridFinishedFuture<>((IgniteInternalTx)null);
      }
  
      /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index fa0f293,5d39822..bc3744e
--- 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
@@@ -360,20 -353,20 +360,20 @@@ public abstract class IgniteTxLocalAdap
          if (!async) {
              try {
                  if (!readThrough || !cacheCtx.readThrough()) {
 -                    for (K key : keys)
 +                    for (KeyCacheObject key : keys)
                          c.apply(key, null);
  
-                     return new GridFinishedFuture<>(cctx.kernalContext(), false);
+                     return new GridFinishedFuture<>(false);
                  }
  
-                 return new GridFinishedFuture<>(cctx.kernalContext(),
+                 return new GridFinishedFuture<>(
                      cacheCtx.store().loadAllFromStore(this, keys, c));
              }
              catch (IgniteCheckedException e) {
-                 return new GridFinishedFuture<>(cctx.kernalContext(), e);
+                 return new GridFinishedFuture<>(e);
              }
          }
 -        else
 +        else {
              return cctx.kernalContext().closure().callLocalSafe(
                  new GPC<Boolean>() {
                      @Override public Boolean call() throws Exception {
@@@ -1408,188 -1411,177 +1408,194 @@@
          if (log.isDebugEnabled())
              log.debug("Loading missed values for missed map: " + missedMap);
  
 -        final Collection<K> loaded = new HashSet<>();
 +        final Collection<KeyCacheObject> loaded = new HashSet<>();
  
-         return new GridEmbeddedFuture<>(cctx.kernalContext(),
-             loadMissing(
-                 cacheCtx,
-                 true, false, missedMap.keySet(), deserializePortable, skipVals, new CI2<KeyCacheObject, Object>() {
-                 /** */
-                 private GridCacheVersion nextVer;
+         return new GridEmbeddedFuture<>(
+             new C2<Boolean, Exception, Map<K, V>>() {
+                 @Override public Map<K, V> apply(Boolean b, Exception e) {
+                     if (e != null) {
+                         setRollbackOnly();
  
-                 @Override public void apply(KeyCacheObject key, Object val) {
-                     if (isRollbackOnly()) {
-                         if (log.isDebugEnabled())
-                             log.debug("Ignoring loaded value for read because transaction was rolled back: " +
-                                 IgniteTxLocalAdapter.this);
+                         throw new GridClosureException(e);
+                     }
+ 
+                     if (!b && !readCommitted()) {
+                         // There is no store - we must mark the entries.
 -                        for (K key : missedMap.keySet()) {
 -                            IgniteTxEntry<K, V> txEntry = entry(cacheCtx.txKey(key));
++                        for (KeyCacheObject key : missedMap.keySet()) {
++                            IgniteTxEntry txEntry = entry(cacheCtx.txKey(key));
  
-                         return;
+                             if (txEntry != null)
+                                 txEntry.markValid();
+                         }
                      }
  
-                     GridCacheVersion ver = missedMap.get(key);
+                     if (readCommitted()) {
 -                        Collection<K> notFound = new HashSet<>(missedMap.keySet());
++                        Collection<KeyCacheObject> notFound = new HashSet<>(missedMap.keySet());
  
-                     if (ver == null) {
-                         if (log.isDebugEnabled())
-                             log.debug("Value from storage was never asked for [key=" + key + ", val=" + val + ']');
+                         notFound.removeAll(loaded);
  
-                         return;
-                     }
+                         // In read-committed mode touch entries that have just been read.
 -                        for (K key : notFound) {
 -                            IgniteTxEntry<K, V> txEntry = entry(cacheCtx.txKey(key));
++                        for (KeyCacheObject key : notFound) {
++                            IgniteTxEntry txEntry = entry(cacheCtx.txKey(key));
  
-                     CacheObject cacheVal = cacheCtx.toCacheObject(val);
 -                            GridCacheEntryEx<K, V> entry = txEntry == null ? cacheCtx.cache().peekEx(key) :
++                            GridCacheEntryEx entry = txEntry == null ? cacheCtx.cache().peekEx(key) :
+                                 txEntry.cached();
  
-                     CacheObject visibleVal = cacheVal;
+                             if (entry != null)
+                                 cacheCtx.evicts().touch(entry, topologyVersion());
+                         }
+                     }
  
-                     IgniteTxKey txKey = cacheCtx.txKey(key);
+                     return map;
+                 }
+             },
+             loadMissing(
+                 cacheCtx,
+                 true,
+                 false,
+                 missedMap.keySet(),
+                 deserializePortable,
+                 skipVals,
 -                new CI2<K, V>() {
++                new CI2<KeyCacheObject, Object>() {
+                     /** */
+                     private GridCacheVersion nextVer;
  
-                     IgniteTxEntry txEntry = entry(txKey);
 -                    @Override public void apply(K key, V val) {
++                    @Override public void apply(KeyCacheObject key, Object val) {
+                         if (isRollbackOnly()) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Ignoring loaded value for read because transaction was rolled back: " +
+                                     IgniteTxLocalAdapter.this);
  
-                     if (txEntry != null) {
-                         if (!readCommitted())
-                             txEntry.readValue(cacheVal);
+                             return;
+                         }
  
-                         if (!F.isEmpty(txEntry.entryProcessors()))
-                             visibleVal = txEntry.applyEntryProcessors(visibleVal);
-                     }
+                         GridCacheVersion ver = missedMap.get(key);
  
-                     // In pessimistic mode we hold the lock, so filter validation
-                     // should always be valid.
-                     if (pessimistic())
-                         ver = null;
+                         if (ver == null) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Value from storage was never asked for [key=" + key + ", val=" + val + ']');
  
-                     // Initialize next version.
-                     if (nextVer == null)
-                         nextVer = cctx.versions().next(topologyVersion());
+                             return;
+                         }
  
-                     while (true) {
-                         assert txEntry != null || readCommitted() || groupLock() || skipVals;
 -                        V visibleVal = val;
++                        CacheObject cacheVal = cacheCtx.toCacheObject(val);
 +
-                         GridCacheEntryEx e = txEntry == null ? entryEx(cacheCtx, txKey) : txEntry.cached();
++                        CacheObject visibleVal = cacheVal;
  
-                         try {
-                             // Must initialize to true since even if filter didn't pass,
-                             // we still record the transaction value.
-                             boolean set;
 -                        IgniteTxKey<K> txKey = cacheCtx.txKey(key);
++                        IgniteTxKey txKey = cacheCtx.txKey(key);
  
-                             try {
-                                 set = e.versionedValue(cacheVal, ver, nextVer);
-                             }
-                             catch (GridCacheEntryRemovedException ignore) {
-                                 if (log.isDebugEnabled())
-                                     log.debug("Got removed entry in transaction getAll method " +
-                                         "(will try again): " + e);
 -                        IgniteTxEntry<K, V> txEntry = entry(txKey);
++                        IgniteTxEntry txEntry = entry(txKey);
  
-                                 if (pessimistic() && !readCommitted() && !isRollbackOnly() &&
-                                     (!groupLock() || F.eq(e.key(), groupLockKey()))) {
-                                     U.error(log, "Inconsistent transaction state (entry got removed while " +
-                                         "holding lock) [entry=" + e + ", tx=" + IgniteTxLocalAdapter.this + "]");
+                         if (txEntry != null) {
+                             if (!readCommitted())
 -                                txEntry.readValue(val);
++                                txEntry.readValue(cacheVal);
  
-                                     setRollbackOnly();
+                             if (!F.isEmpty(txEntry.entryProcessors()))
+                                 visibleVal = txEntry.applyEntryProcessors(visibleVal);
+                         }
  
-                                     return;
-                                 }
+                         // In pessimistic mode we hold the lock, so filter validation
+                         // should always be valid.
+                         if (pessimistic())
+                             ver = null;
  
-                                 if (txEntry != null)
-                                     txEntry.cached(entryEx(cacheCtx, txKey));
+                         // Initialize next version.
+                         if (nextVer == null)
+                             nextVer = cctx.versions().next(topologyVersion());
  
-                                 continue; // While loop.
-                             }
+                         while (true) {
+                             assert txEntry != null || readCommitted() || groupLock() || skipVals;
  
-                             // In pessimistic mode, we should always be able to set.
-                             assert set || !pessimistic();
 -                            GridCacheEntryEx<K, V> e = txEntry == null ? entryEx(cacheCtx, txKey) : txEntry.cached();
++                            GridCacheEntryEx e = txEntry == null ? entryEx(cacheCtx, txKey) : txEntry.cached();
  
-                             if (readCommitted() || groupLock() || skipVals) {
-                                 cacheCtx.evicts().touch(e, topologyVersion());
+                             try {
+                                 // Must initialize to true since even if filter didn't pass,
+                                 // we still record the transaction value.
+                                 boolean set;
  
-                                 if (visibleVal != null) {
-                                     cacheCtx.addResult(map,
-                                         key,
-                                         visibleVal,
-                                         skipVals,
-                                         keepCacheObjects,
-                                         deserializePortable,
-                                         false);
+                                 try {
 -                                    set = e.versionedValue(val, ver, nextVer);
++                                    set = e.versionedValue(cacheVal, ver, nextVer);
                                  }
-                             }
-                             else {
-                                 assert txEntry != null;
+                                 catch (GridCacheEntryRemovedException ignore) {
+                                     if (log.isDebugEnabled())
+                                         log.debug("Got removed entry in transaction getAll method " +
+                                             "(will try again): " + e);
  
-                                 txEntry.setAndMarkValid(cacheVal);
+                                     if (pessimistic() && !readCommitted() && !isRollbackOnly() &&
+                                         (!groupLock() || F.eq(e.key(), groupLockKey()))) {
+                                         U.error(log, "Inconsistent transaction state (entry got removed while " +
+                                             "holding lock) [entry=" + e + ", tx=" + IgniteTxLocalAdapter.this + "]");
  
-                                 if (visibleVal != null) {
-                                     cacheCtx.addResult(map,
-                                         key,
-                                         visibleVal,
-                                         skipVals,
-                                         keepCacheObjects,
-                                         deserializePortable,
-                                         false);
-                                 }
-                             }
+                                         setRollbackOnly();
  
-                             loaded.add(key);
+                                         return;
+                                     }
  
-                             if (log.isDebugEnabled())
-                                 log.debug("Set value loaded from store into entry from transaction [set=" + set +
-                                     ", matchVer=" + ver + ", newVer=" + nextVer + ", entry=" + e + ']');
+                                     if (txEntry != null)
+                                         txEntry.cached(entryEx(cacheCtx, txKey), txEntry.keyBytes());
  
-                             break; // While loop.
-                         }
-                         catch (IgniteCheckedException ex) {
-                             throw new IgniteException("Failed to put value for cache entry: " + e, ex);
-                         }
-                     }
-                 }
-             }),
-             new C2<Boolean, Exception, Map<K, V>>() {
-                 @Override public Map<K, V> apply(Boolean b, Exception e) {
-                     if (e != null) {
-                         setRollbackOnly();
+                                     continue; // While loop.
+                                 }
  
-                         throw new GridClosureException(e);
-                     }
+                                 // In pessimistic mode, we should always be able to set.
+                                 assert set || !pessimistic();
  
-                     if (!b && !readCommitted()) {
-                         // There is no store - we must mark the entries.
-                         for (KeyCacheObject key : missedMap.keySet()) {
-                             IgniteTxEntry txEntry = entry(cacheCtx.txKey(key));
+                                 if (readCommitted() || groupLock() || skipVals) {
+                                     cacheCtx.evicts().touch(e, topologyVersion());
  
-                             if (txEntry != null)
-                                 txEntry.markValid();
-                         }
-                     }
 -                                    if (visibleVal != null)
 -                                        map.put(key, (V)CU.skipValue(visibleVal, skipVals));
++                                    if (visibleVal != null) {
++                                        cacheCtx.addResult(map,
++                                            key,
++                                            visibleVal,
++                                            skipVals,
++                                            keepCacheObjects,
++                                            deserializePortable,
++                                            false);
++                                    }
+                                 }
+                                 else {
+                                     assert txEntry != null;
  
-                     if (readCommitted()) {
-                         Collection<KeyCacheObject> notFound = new HashSet<>(missedMap.keySet());
 -                                    txEntry.setAndMarkValid(val);
++                                    txEntry.setAndMarkValid(cacheVal);
  
-                         notFound.removeAll(loaded);
 -                                    if (visibleVal != null)
 -                                        map.put(key, visibleVal);
++                                    if (visibleVal != null) {
++                                        cacheCtx.addResult(map,
++                                            key,
++                                            visibleVal,
++                                            skipVals,
++                                            keepCacheObjects,
++                                            deserializePortable,
++                                            false);
++                                    }
+                                 }
  
-                         // In read-committed mode touch entries that have just been read.
-                         for (KeyCacheObject key : notFound) {
-                             IgniteTxEntry txEntry = entry(cacheCtx.txKey(key));
+                                 loaded.add(key);
  
-                             GridCacheEntryEx entry = txEntry == null ? cacheCtx.cache().peekEx(key) :
-                                 txEntry.cached();
+                                 if (log.isDebugEnabled())
+                                     log.debug("Set value loaded from store into entry from transaction [set=" + set +
+                                         ", matchVer=" + ver + ", newVer=" + nextVer + ", entry=" + e + ']');
  
-                             if (entry != null)
-                                 cacheCtx.evicts().touch(entry, topologyVersion());
+                                 break; // While loop.
+                             }
+                             catch (IgniteCheckedException ex) {
+                                 throw new IgniteException("Failed to put value for cache entry: " + e, ex);
+                             }
                          }
                      }
- 
-                     return map;
-                 }
-             });
+                 })
+         );
      }
  
      /** {@inheritDoc} */
 -    @Override public IgniteInternalFuture<Map<K, V>> getAllAsync(
 -        final GridCacheContext<K, V> cacheCtx,
 -        Collection<? extends K> keys,
 -        @Nullable GridCacheEntryEx<K, V> cached,
 +    @Override public <K, V> IgniteInternalFuture<Map<K, V>> getAllAsync(
 +        final GridCacheContext cacheCtx,
 +        Collection<KeyCacheObject> keys,
 +        @Nullable GridCacheEntryEx cached,
          final boolean deserializePortable,
 -        final boolean skipVals) {
 +        final boolean skipVals,
 +        final boolean keepCacheObjects) {
          if (F.isEmpty(keys))
-             return new GridFinishedFuture<>(cctx.kernalContext(), Collections.<K, V>emptyMap());
+             return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
  
          init();
  
@@@ -1616,11 -1608,10 +1622,11 @@@
                  missed,
                  keysCnt,
                  deserializePortable,
 -                skipVals);
 +                skipVals,
 +                keepCacheObjects);
  
              if (single && missed.isEmpty())
-                 return new GridFinishedFuture<>(cctx.kernalContext(), retMap);
+                 return new GridFinishedFuture<>(retMap);
  
              // Handle locks.
              if (pessimistic() && !readCommitted() && !groupLock() && !skipVals) {
@@@ -1730,17 -1715,10 +1736,17 @@@
                              }
                          }
  
 -                        if (!missed.isEmpty() && (cacheCtx.isReplicated() || cacheCtx.isLocal()))
 -                            return checkMissed(cacheCtx, retMap, missed, null, deserializePortable, skipVals);
 +                        if (!missed.isEmpty() && (cacheCtx.isReplicated() || cacheCtx.isLocal())) {
 +                            return checkMissed(cacheCtx,
 +                                retMap,
 +                                missed,
 +                                null,
 +                                deserializePortable,
 +                                skipVals,
 +                                keepCacheObjects);
 +                        }
  
-                         return new GridFinishedFuture<>(cctx.kernalContext(), Collections.<K, V>emptyMap());
+                         return new GridFinishedFuture<>(Collections.<K, V>emptyMap());
                      }
                  };
  
@@@ -1787,23 -1764,16 +1792,22 @@@
  
                  if (!missed.isEmpty()) {
                      if (!readCommitted())
 -                        for (Iterator<K> it = missed.keySet().iterator(); it.hasNext(); )
 -                            if (retMap.containsKey(it.next()))
 +                        for (Iterator<KeyCacheObject> it = missed.keySet().iterator(); it.hasNext(); ) {
 +                            KeyCacheObject cacheKey = it.next();
 +
 +                            K keyVal =
 +                                (K)(keepCacheObjects ? cacheKey : cacheKey.value(cacheCtx.cacheObjectContext(), false));
 +
 +                            if (retMap.containsKey(keyVal))
                                  it.remove();
 +                        }
  
                      if (missed.isEmpty())
-                         return new GridFinishedFuture<>(cctx.kernalContext(), retMap);
+                         return new GridFinishedFuture<>(retMap);
  
                      return new GridEmbeddedFuture<>(
-                         cctx.kernalContext(),
                          // First future.
 -                        checkMissed(cacheCtx, retMap, missed, redos, deserializePortable, skipVals),
 +                        checkMissed(cacheCtx, retMap, missed, redos, deserializePortable, skipVals, keepCacheObjects),
                          // Closure that returns another future, based on result from first.
                          new PMC<Map<K, V>>() {
                              @Override public IgniteInternalFuture<Map<K, V>> postMiss(Map<K, V> map) {
@@@ -1981,10 -1938,10 +1985,10 @@@
              addActiveCache(cacheCtx);
          }
          catch (IgniteCheckedException e) {
-             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+             return new GridFinishedFuture<>(e);
          }
  
 -        Set<K> skipped = null;
 +        Set<KeyCacheObject> skipped = null;
  
          boolean rmv = lookup == null && invokeMap == null;
  
@@@ -2288,10 -2244,8 +2292,8 @@@
                  });
  
              return new GridEmbeddedFuture<>(
-                 cctx.kernalContext(),
-                 fut,
 -                new C2<Boolean, Exception, Set<K>>() {
 -                    @Override public Set<K> apply(Boolean b, Exception e) {
 +                new C2<Boolean, Exception, Set<KeyCacheObject>>() {
 +                    @Override public Set<KeyCacheObject> apply(Boolean b, Exception e) {
                          if (e != null)
                              throw new GridClosureException(e);
  
@@@ -2670,11 -2661,11 +2672,11 @@@
                          loadFut.get();
                      }
                      catch (IgniteCheckedException e) {
-                         return new GridFinishedFutureEx<>(new GridCacheReturn(localResult()), e);
+                         return new GridFinishedFuture<>(e);
                      }
  
 -                    return commitAsync().chain(new CX1<IgniteInternalFuture<IgniteInternalTx>, GridCacheReturn<V>>() {
 -                        @Override public GridCacheReturn<V> applyx(IgniteInternalFuture<IgniteInternalTx> txFut) throws IgniteCheckedException {
 +                    return commitAsync().chain(new CX1<IgniteInternalFuture<IgniteInternalTx>, GridCacheReturn>() {
 +                        @Override public GridCacheReturn applyx(IgniteInternalFuture<IgniteInternalTx> txFut) throws IgniteCheckedException {
                              txFut.get();
  
                              return implicitRes;
@@@ -2752,10 -2759,10 +2754,10 @@@
              checkValid();
          }
          catch (IgniteCheckedException e) {
-             return new GridFinishedFuture<>(cctx.kernalContext(), e);
+             return new GridFinishedFuture<>(e);
          }
  
 -        final GridCacheReturn<V> ret = new GridCacheReturn<>(false);
 +        final GridCacheReturn ret = new GridCacheReturn(localResult(), false);
  
          if (F.isEmpty(keys0)) {
              if (implicit()) {
@@@ -3009,8 -3006,8 +3011,8 @@@
                      isolation,
                      isInvalidate(),
                      -1L,
 -                    CU.<K, V>empty()) :
 +                    CU.empty0()) :
-                 new GridFinishedFuture<>(cctx.kernalContext());
+                 new GridFinishedFuture<>();
          }
          catch (IgniteCheckedException e) {
              setRollbackOnly();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index b2329bb,609bd3e..71098ee
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@@ -65,25 -65,25 +65,25 @@@ public class IgniteTxManager extends Gr
      private static final int TX_SALVAGE_TIMEOUT = Integer.getInteger(IGNITE_TX_SALVAGE_TIMEOUT, 100);
  
      /** Committing transactions. */
-     private final ThreadLocal<IgniteInternalTx> threadCtx = new GridThreadLocalEx<>();
+     private final ThreadLocal<IgniteInternalTx> threadCtx = new ThreadLocal<>();
  
      /** Per-thread transaction map. */
 -    private final ConcurrentMap<Long, IgniteInternalTx<K, V>> threadMap = newMap();
 +    private final ConcurrentMap<Long, IgniteInternalTx> threadMap = newMap();
  
      /** Per-ID map. */
 -    private final ConcurrentMap<GridCacheVersion, IgniteInternalTx<K, V>> idMap = newMap();
 +    private final ConcurrentMap<GridCacheVersion, IgniteInternalTx> idMap = newMap();
  
      /** Per-ID map for near transactions. */
 -    private final ConcurrentMap<GridCacheVersion, IgniteInternalTx<K, V>> nearIdMap = newMap();
 +    private final ConcurrentMap<GridCacheVersion, IgniteInternalTx> nearIdMap = newMap();
  
      /** TX handler. */
 -    private IgniteTxHandler<K, V> txHandler;
 +    private IgniteTxHandler txHandler;
  
      /** All transactions. */
 -    private final Queue<IgniteInternalTx<K, V>> committedQ = new ConcurrentLinkedDeque8<>();
 +    private final Queue<IgniteInternalTx> committedQ = new ConcurrentLinkedDeque8<>();
  
      /** Preparing transactions. */
 -    private final Queue<IgniteInternalTx<K, V>> prepareQ = new ConcurrentLinkedDeque8<>();
 +    private final Queue<IgniteInternalTx> prepareQ = new ConcurrentLinkedDeque8<>();
  
      /** Minimum start version. */
      private final ConcurrentNavigableMap<GridCacheVersion, AtomicInt> startVerCnts =
@@@ -389,11 -389,11 +389,11 @@@
       * @param tx Created transaction.
       * @return Started transaction.
       */
 -    @Nullable public <T extends IgniteInternalTx<K, V>> T onCreated(T tx) {
 -        ConcurrentMap<GridCacheVersion, IgniteInternalTx<K, V>> txIdMap = transactionMap(tx);
 +    @Nullable public <T extends IgniteInternalTx> T onCreated(T tx) {
 +        ConcurrentMap<GridCacheVersion, IgniteInternalTx> txIdMap = transactionMap(tx);
  
          // Start clean.
-         txContextReset();
+         resetContext();
  
          if (isCompleted(tx)) {
              if (log.isDebugEnabled())

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyImpl.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessor.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataLoaderImpl.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataLoaderImpl.java
index 81fdbed,90f519d..193e3f8
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataLoaderImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataLoaderImpl.java
@@@ -169,8 -175,13 +169,11 @@@ public class IgniteDataLoaderImpl<K, V
          assert ctx != null;
  
          this.ctx = ctx;
 -        this.cacheName = cacheName;
 -        this.flushQ = flushQ;
 -        this.compact = compact;
 +        this.cacheObjProc = ctx.cacheObjects();
  
+         if (log == null)
+             log = U.logger(ctx, logRef, IgniteDataLoaderImpl.class);
+ 
          ClusterNode node = F.first(ctx.grid().cluster().forCacheNodes(cacheName).nodes());
  
          if (node == null)
@@@ -458,13 -397,8 +461,13 @@@
  
              return new IgniteFutureImpl<>(resFut);
          }
 -        catch (IgniteException e) {
 +        catch (Throwable e) {
 +            resFut.onDone(e);
 +
 +            if (e instanceof Error)
 +                throw e;
 +
-             return new IgniteFinishedFutureImpl<>(ctx, e);
+             return new IgniteFinishedFutureImpl<>(e);
          }
          finally {
              leaveBusy();
@@@ -941,9 -870,9 +944,9 @@@
              synchronized (this) {
                  curFut0 = curFut;
  
-                 curFut0.listenAsync(lsnr);
+                 curFut0.listen(lsnr);
  
 -                for (Map.Entry<K, V> entry : newEntries)
 +                for (IgniteDataLoaderEntry entry : newEntries)
                      entries.add(entry);
  
                  if (entries.size() >= bufSize) {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheFinishPartitionsSelfTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePutAllFailoverSelfTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheEventAbstractTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e59ea498/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
----------------------------------------------------------------------


Mime
View raw message