ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [04/37] incubator-ignite git commit: ignite-43
Date Tue, 20 Jan 2015 08:49:36 GMT
ignite-43


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/13740b70
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/13740b70
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/13740b70

Branch: refs/heads/sprint-1
Commit: 13740b7095469e54b1748dd4914283a0b2a2304b
Parents: f9f0231
Author: sboikov <sboikov@gridgain.com>
Authored: Mon Jan 12 11:45:06 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Mon Jan 12 17:21:01 2015 +0300

----------------------------------------------------------------------
 .../grid/kernal/GridEventConsumeHandler.java    |   2 +-
 .../processors/cache/GridCacheMapEntry.java     |  21 +-
 .../GridCacheDataStructuresManager.java         |   5 +-
 .../GridCacheContinuousQueryAdapter.java        |  14 +-
 .../GridCacheContinuousQueryHandler.java        |  23 +-
 .../GridCacheContinuousQueryHandlerV2.java      |   4 +-
 .../GridCacheContinuousQueryListener.java       |   3 +-
 .../GridCacheContinuousQueryManager.java        | 110 +++--
 .../continuous/GridContinuousMessage.java       |  52 +-
 .../continuous/GridContinuousMessageType.java   |   5 +-
 .../continuous/GridContinuousProcessor.java     | 181 ++++++-
 .../service/GridServiceProcessor.java           |   4 +-
 .../IgniteCacheEntryListenerAbstractTest.java   | 482 ++++++++++++++++---
 ...eCacheEntryListenerEagerTtlDisabledTest.java |  22 +
 .../junits/common/GridCommonAbstractTest.java   |  13 +
 .../bamboo/GridDataGridTestSuite.java           |   1 +
 16 files changed, 772 insertions(+), 170 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
index a48f6e1..778b353 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/GridEventConsumeHandler.java
@@ -142,7 +142,7 @@ class GridEventConsumeHandler implements GridContinuousHandler {
                                     }
                                 }
 
-                                ctx.continuous().addNotification(nodeId, routineId, wrapper, null);
+                                ctx.continuous().addNotification(nodeId, routineId, wrapper, null, false);
                             }
                             catch (IgniteCheckedException e) {
                                 U.error(ctx.log(getClass()), "Failed to send event notification to node: " + nodeId, e);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
index b19e44f..aee3806 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
@@ -3342,12 +3342,23 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
                             obsolete = true; // Success, will return "true".
                     }
 
-                    if (cctx.events().isRecordable(EVT_CACHE_OBJECT_EXPIRED))
-                        cctx.events().addEvent(partition(), key, cctx.localNodeId(), null, EVT_CACHE_OBJECT_EXPIRED,
-                            null, false, expiredVal, expiredVal != null || hasOldBytes, null, null, null);
-                }
-
+                    if (cctx.events().isRecordable(EVT_CACHE_OBJECT_EXPIRED)) {
+                        cctx.events().addEvent(partition(),
+                            key,
+                            cctx.localNodeId(),
+                            null,
+                            EVT_CACHE_OBJECT_EXPIRED,
+                            null,
+                            false,
+                            expiredVal,
+                            expiredVal != null || hasOldBytes,
+                            null,
+                            null,
+                            null);
+                    }
 
+                    cctx.continuousQueries().onEntryExpired(this, key, expiredVal, null);
+                }
             }
         }
         catch (IgniteCheckedException e) {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
index db072d6..384329b 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheDataStructuresManager.java
@@ -679,7 +679,10 @@ public final class GridCacheDataStructuresManager<K, V> extends GridCacheManager
                }
             });
 
-            queueQry.execute(cctx.isLocal() || cctx.isReplicated() ? cctx.grid().forLocal() : null, true, false);
+            queueQry.execute(cctx.isLocal() || cctx.isReplicated() ? cctx.grid().forLocal() : null,
+                true,
+                false,
+                false);
         }
 
         GridCacheQueueProxy queue = queuesMap.get(header.id());

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
index c00c961..06eb2a8 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryAdapter.java
@@ -213,12 +213,12 @@ public class GridCacheContinuousQueryAdapter<K, V> implements GridCacheContinuou
 
     /** {@inheritDoc} */
     @Override public void execute() throws IgniteCheckedException {
-        execute(null, false, false);
+        execute(null, false, false, false);
     }
 
     /** {@inheritDoc} */
     @Override public void execute(@Nullable ClusterGroup prj) throws IgniteCheckedException {
-        execute(prj, false, false);
+        execute(prj, false, false, false);
     }
 
     /**
@@ -227,9 +227,13 @@ public class GridCacheContinuousQueryAdapter<K, V> implements GridCacheContinuou
      * @param prj Grid projection.
      * @param internal If {@code true} then query notified about internal entries updates.
      * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
+     * @param sync {@code True} if query created for synchronous {@link CacheEntryListener}.
      * @throws IgniteCheckedException If failed.
      */
-    public void execute(@Nullable ClusterGroup prj, boolean internal, boolean entryLsnr) throws IgniteCheckedException {
+    public void execute(@Nullable ClusterGroup prj,
+        boolean internal,
+        boolean entryLsnr,
+        boolean sync) throws IgniteCheckedException {
         if (locCb == null)
             throw new IllegalStateException("Mandatory local callback is not set for the query: " + this);
 
@@ -282,6 +286,7 @@ public class GridCacheContinuousQueryAdapter<K, V> implements GridCacheContinuou
                     prjPred,
                     internal,
                     entryLsnr,
+                    sync,
                     ctx.kernalContext().job().currentTaskNameHash());
             }
             else {
@@ -291,7 +296,8 @@ public class GridCacheContinuousQueryAdapter<K, V> implements GridCacheContinuou
                     rmtFilter,
                     prjPred,
                     internal,
-                    entryLsnr);
+                    entryLsnr,
+                    sync);
             }
 
             routineId = ctx.kernalContext().continuous().startRoutine(hnd,

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
index e61b2a2..9162b89 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandler.java
@@ -65,6 +65,9 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
     /** Entry listener flag. */
     private boolean entryLsnr;
 
+    /** Synchronous listener flag. */
+    private boolean sync;
+
     /**
      * Required by {@link Externalizable}.
      */
@@ -80,6 +83,7 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
      * @param prjPred Projection predicate.
      * @param internal If {@code true} then query is notified about internal entries updates.
      * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
+     * @param sync {@code True} if query created for synchronous {@link CacheEntryListener}.
      */
     GridCacheContinuousQueryHandler(@Nullable String cacheName,
         Object topic,
@@ -87,7 +91,8 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
         @Nullable IgnitePredicate<org.gridgain.grid.cache.query.GridCacheContinuousQueryEntry<K, V>> filter,
         @Nullable IgnitePredicate<GridCacheEntry<K, V>> prjPred,
         boolean internal,
-        boolean entryLsnr) {
+        boolean entryLsnr,
+        boolean sync) {
         assert topic != null;
         assert cb != null;
 
@@ -98,6 +103,7 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
         this.prjPred = prjPred;
         this.internal = internal;
         this.entryLsnr = entryLsnr;
+        this.sync = sync;
     }
 
     /** {@inheritDoc} */
@@ -150,7 +156,9 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
                 }
             }
 
-            @Override public void onEntryUpdate(GridCacheContinuousQueryEntry<K, V> e, boolean recordEvt) {
+            @Override public void onEntryUpdate(GridCacheContinuousQueryEntry<K, V> e,
+                boolean recordEvt,
+                boolean sync) {
                 boolean notify;
 
                 GridCacheFlag[] f = cacheContext(ctx).forceLocalRead();
@@ -185,7 +193,7 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
                                 depMgr.prepare(e);
                             }
 
-                            ctx.continuous().addNotification(nodeId, routineId, e, topic);
+                            ctx.continuous().addNotification(nodeId, routineId, e, topic, sync);
                         }
                         catch (IgniteCheckedException ex) {
                             U.error(ctx.log(getClass()), "Failed to send event notification to node: " + nodeId, ex);
@@ -250,12 +258,13 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
             }
         };
 
-        return manager(ctx).registerListener(routineId, lsnr, internal, entryLsnr);
+        return manager(ctx).registerListener(routineId, lsnr, internal, entryLsnr, sync);
     }
 
     /** {@inheritDoc} */
     @Override public void onListenerRegistered(UUID routineId, GridKernalContext ctx) {
-        manager(ctx).iterate(internal, routineId, entryLsnr);
+        if (!entryLsnr)
+            manager(ctx).iterate(internal, routineId);
     }
 
     /** {@inheritDoc} */
@@ -382,6 +391,8 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
         out.writeBoolean(internal);
 
         out.writeBoolean(entryLsnr);
+
+        out.writeBoolean(sync);
     }
 
     /** {@inheritDoc} */
@@ -407,6 +418,8 @@ class GridCacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
         internal = in.readBoolean();
 
         entryLsnr = in.readBoolean();
+
+        sync = in.readBoolean();
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
index 63209ec..b4216d5 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryHandlerV2.java
@@ -43,6 +43,7 @@ public class GridCacheContinuousQueryHandlerV2<K, V> extends GridCacheContinuous
      * @param prjPred Projection predicate.
      * @param internal If {@code true} then query is notified about internal entries updates.
      * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
+     * @param sync {@code True} if query created for synchronous {@link CacheEntryListener}.
      * @param taskHash Task hash.
      */
     public GridCacheContinuousQueryHandlerV2(@Nullable String cacheName,
@@ -52,8 +53,9 @@ public class GridCacheContinuousQueryHandlerV2<K, V> extends GridCacheContinuous
         @Nullable IgnitePredicate<GridCacheEntry<K, V>> prjPred,
         boolean internal,
         boolean entryLsnr,
+        boolean sync,
         int taskHash) {
-        super(cacheName, topic, cb, filter, prjPred, internal, entryLsnr);
+        super(cacheName, topic, cb, filter, prjPred, internal, entryLsnr, sync);
 
         this.taskHash = taskHash;
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryListener.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryListener.java
index dd6b5f9..2707428 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryListener.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryListener.java
@@ -23,6 +23,7 @@ interface GridCacheContinuousQueryListener<K, V> {
      *
      * @param e Entry.
      * @param recordEvt Whether to record event.
+     * @param sync {@code True} if event is synchronous.
      */
-    public void onEntryUpdate(GridCacheContinuousQueryEntry<K, V> e, boolean recordEvt);
+    public void onEntryUpdate(GridCacheContinuousQueryEntry<K, V> e, boolean recordEvt, boolean sync);
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
index 8c1b70e..8bcbceb 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java
@@ -185,21 +185,23 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
         if (F.isEmpty(lsnrCol))
             return;
 
-        GridCacheContinuousQueryEntry<K, V> e0 = new GridCacheContinuousQueryEntry<>(
-            cctx,
-            e.wrap(false),
-            key,
-            null,
-            null,
-            oldVal,
-            oldBytes,
-            true);
-
-        for (ListenerInfo<K, V> lsnr : lsnrCol.values()) {
-            if (!lsnr.entryListener())
-                continue;
-
-            lsnr.onEntryUpdate(e0, false);
+        if (cctx.isReplicated() || cctx.affinity().primary(cctx.localNode(), key, -1)) {
+            GridCacheContinuousQueryEntry<K, V> e0 = new GridCacheContinuousQueryEntry<>(
+                cctx,
+                e.wrap(false),
+                key,
+                null,
+                null,
+                oldVal,
+                oldBytes,
+                true);
+
+            for (ListenerInfo<K, V> lsnr : lsnrCol.values()) {
+                if (!lsnr.entryListener())
+                    continue;
+
+                lsnr.onEntryUpdate(e0, false);
+            }
         }
     }
 
@@ -252,7 +254,7 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
 
             qry.remoteFilter(fltr);
 
-            qry.execute(null, false, true);
+            qry.execute(null, false, true, lsnrCfg.isSynchronous());
         }
         catch (IgniteCheckedException e) {
             lsnrQrys.remove(lsnrCfg, qry); // Remove query if failed to execute it.
@@ -279,12 +281,17 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
      * @param lsnr Listener.
      * @param internal Internal flag.
      * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
+     * @param sync {@code True} if query created for synchronous {@link CacheEntryListener}.
      * @return Whether listener was actually registered.
      */
-    boolean registerListener(UUID lsnrId, GridCacheContinuousQueryListener<K, V> lsnr,
+    boolean registerListener(UUID lsnrId,
+        GridCacheContinuousQueryListener<K, V> lsnr,
         boolean internal,
-        boolean entryLsnr) {
-        ListenerInfo<K, V> info = new ListenerInfo<>(lsnr, entryLsnr);
+        boolean entryLsnr,
+        boolean sync) {
+        assert !sync || entryLsnr;
+
+        ListenerInfo<K, V> info = new ListenerInfo<>(lsnr, entryLsnr, sync);
 
         boolean added;
 
@@ -326,36 +333,33 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
      * Iterates through existing data.
      *
      * @param internal Internal flag.
-     * @param entryLsnr {@code True} if query created for {@link CacheEntryListener}.
      * @param id Listener ID.
      */
-    void iterate(boolean internal, UUID id, boolean entryLsnr) {
+    void iterate(boolean internal, UUID id) {
         ListenerInfo<K, V> info = internal ? intLsnrs.get(id) : lsnrs.get(id);
 
         assert info != null;
 
-        if (!entryLsnr) {
-            Set<GridCacheEntry<K, V>> entries;
-
-            if (cctx.isReplicated())
-                entries = internal ? cctx.cache().entrySetx() :
-                    cctx.cache().entrySet();
-            else
-                entries = internal ? cctx.cache().primaryEntrySetx() :
-                    cctx.cache().primaryEntrySet();
-
-            for (GridCacheEntry<K, V> e : entries) {
-                GridCacheContinuousQueryEntry<K, V> qryEntry = new GridCacheContinuousQueryEntry<>(cctx,
-                    e,
-                    e.getKey(),
-                    e.getValue(),
-                    null,
-                    null,
-                    null,
-                    false);
-
-                info.onIterate(qryEntry, !internal && cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ));
-            }
+        Set<GridCacheEntry<K, V>> entries;
+
+        if (cctx.isReplicated())
+            entries = internal ? cctx.cache().entrySetx() :
+                cctx.cache().entrySet();
+        else
+            entries = internal ? cctx.cache().primaryEntrySetx() :
+                cctx.cache().primaryEntrySet();
+
+        for (GridCacheEntry<K, V> e : entries) {
+            GridCacheContinuousQueryEntry<K, V> qryEntry = new GridCacheContinuousQueryEntry<>(cctx,
+                e,
+                e.getKey(),
+                e.getValue(),
+                null,
+                null,
+                null,
+                false);
+
+            info.onIterate(qryEntry, !internal && cctx.gridEvents().isRecordable(EVT_CACHE_QUERY_OBJECT_READ));
         }
 
         info.flushPending();
@@ -369,18 +373,26 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
         private final GridCacheContinuousQueryListener<K, V> lsnr;
 
         /** Pending entries. */
-        private Collection<PendingEntry<K, V>> pending = new LinkedList<>();
+        private Collection<PendingEntry<K, V>> pending;
+
+        /** */
+        private final boolean entryLsnr;
 
         /** */
-        private boolean entryLsnr;
+        private final boolean sync;
 
         /**
          * @param lsnr Listener.
          * @param entryLsnr {@code True} if listener created for {@link CacheEntryListener}.
+         * @param sync {@code True} if listener is synchronous.
          */
-        private ListenerInfo(GridCacheContinuousQueryListener<K, V> lsnr, boolean entryLsnr) {
+        private ListenerInfo(GridCacheContinuousQueryListener<K, V> lsnr, boolean entryLsnr, boolean sync) {
             this.lsnr = lsnr;
             this.entryLsnr = entryLsnr;
+            this.sync = sync;
+
+            if (!entryLsnr)
+                pending = new LinkedList<>();
         }
 
         /**
@@ -399,7 +411,7 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
             }
 
             if (notifyLsnr)
-                lsnr.onEntryUpdate(e, recordEvt);
+                lsnr.onEntryUpdate(e, recordEvt, sync);
         }
 
         /**
@@ -407,7 +419,7 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
          * @param recordEvt Whether to record event.
          */
         void onIterate(GridCacheContinuousQueryEntry<K, V> e, boolean recordEvt) {
-            lsnr.onEntryUpdate(e, recordEvt);
+            lsnr.onEntryUpdate(e, recordEvt, sync);
         }
 
         /**
@@ -423,7 +435,7 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt
             }
 
             for (PendingEntry<K, V> e : pending0)
-                lsnr.onEntryUpdate(e.entry, e.recordEvt);
+                lsnr.onEntryUpdate(e.entry, e.recordEvt, sync);
         }
 
         /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessage.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessage.java
index 09b7171..27a6429 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessage.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessage.java
@@ -9,6 +9,7 @@
 
 package org.gridgain.grid.kernal.processors.continuous;
 
+import org.apache.ignite.lang.*;
 import org.gridgain.grid.kernal.*;
 import org.gridgain.grid.util.direct.*;
 import org.gridgain.grid.util.typedef.internal.*;
@@ -19,6 +20,8 @@ import java.io.*;
 import java.nio.*;
 import java.util.*;
 
+import static org.gridgain.grid.kernal.processors.continuous.GridContinuousMessageType.*;
+
 /**
  * Continuous processor message.
  */
@@ -40,6 +43,9 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
     /** Serialized message data. */
     private byte[] dataBytes;
 
+    /** Future ID for synchronous event notifications. */
+    private IgniteUuid futId;
+
     /**
      * Required by {@link Externalizable}.
      */
@@ -50,14 +56,19 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
     /**
      * @param type Message type.
      * @param routineId Consume ID.
+     * @param futId Future ID.
      * @param data Optional message data.
      */
-    GridContinuousMessage(GridContinuousMessageType type, UUID routineId, @Nullable Object data) {
+    GridContinuousMessage(GridContinuousMessageType type,
+        @Nullable UUID routineId,
+        @Nullable IgniteUuid futId,
+        @Nullable Object data) {
         assert type != null;
-        assert routineId != null;
+        assert routineId != null || type == MSG_EVT_ACK;
 
         this.type = type;
         this.routineId = routineId;
+        this.futId = futId;
         this.data = data;
     }
 
@@ -78,6 +89,7 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
     /**
      * @return Message data.
      */
+    @SuppressWarnings("unchecked")
     public <T> T data() {
         return (T)data;
     }
@@ -103,14 +115,21 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
         this.dataBytes = dataBytes;
     }
 
+    /**
+     * @return Future ID for synchronous event notification.
+     */
+    @Nullable public IgniteUuid futureId() {
+        return futId;
+    }
+
     /** {@inheritDoc} */
     @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors"})
     @Override public GridTcpCommunicationMessageAdapter clone() {
-        GridContinuousMessage clone = new GridContinuousMessage();
+        GridContinuousMessage _clone = new GridContinuousMessage();
 
-        clone0(clone);
+        clone0(_clone);
 
-        return clone;
+        return _clone;
     }
 
     /** {@inheritDoc} */
@@ -119,6 +138,7 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
 
         clone.type = type;
         clone.routineId = routineId;
+        clone.futId = futId;
         clone.data = data;
         clone.dataBytes = dataBytes;
     }
@@ -143,12 +163,18 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
                 commState.idx++;
 
             case 1:
-                if (!commState.putUuid(routineId))
+                if (!commState.putGridUuid(futId))
                     return false;
 
                 commState.idx++;
 
             case 2:
+                if (!commState.putUuid(routineId))
+                    return false;
+
+                commState.idx++;
+
+            case 3:
                 if (!commState.putEnum(type))
                     return false;
 
@@ -176,6 +202,16 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
                 commState.idx++;
 
             case 1:
+                IgniteUuid futId0 = commState.getGridUuid();
+
+                if (futId0 == GRID_UUID_NOT_READ)
+                    return false;
+
+                futId = futId0;
+
+                commState.idx++;
+
+            case 2:
                 UUID routineId0 = commState.getUuid();
 
                 if (routineId0 == UUID_NOT_READ)
@@ -185,13 +221,13 @@ public class GridContinuousMessage extends GridTcpCommunicationMessageAdapter {
 
                 commState.idx++;
 
-            case 2:
+            case 3:
                 if (buf.remaining() < 1)
                     return false;
 
                 byte type0 = commState.getByte();
 
-                type = GridContinuousMessageType.fromOrdinal(type0);
+                type = fromOrdinal(type0);
 
                 commState.idx++;
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessageType.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessageType.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessageType.java
index 4ebc33a..b07a6e8 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessageType.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousMessageType.java
@@ -28,7 +28,10 @@ enum GridContinuousMessageType {
     MSG_STOP_ACK,
 
     /** Remote event notification. */
-    MSG_EVT_NOTIFICATION;
+    MSG_EVT_NOTIFICATION,
+
+    /** Event notification acknowledgement for synchronous events. */
+    MSG_EVT_ACK;
 
     /** Enumerated values. */
     private static final GridContinuousMessageType[] VALS = values();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousProcessor.java
index 14d89f3..e7f8619 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/continuous/GridContinuousProcessor.java
@@ -69,6 +69,9 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     /** Pending start requests. */
     private final Map<UUID, Collection<GridContinuousMessage>> pending = new HashMap<>();
 
+    /** */
+    private final ConcurrentMap<IgniteUuid, SyncMessageAckFuture> syncMsgFuts = new ConcurrentHashMap8<>();
+
     /** Stopped IDs. */
     private final Collection<UUID> stopped = new HashSet<>();
 
@@ -206,6 +209,21 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                                 unregisterRemote(routineId);
                         }
 
+                        for (Map.Entry<IgniteUuid, SyncMessageAckFuture> e : syncMsgFuts.entrySet()) {
+                            SyncMessageAckFuture fut = e.getValue();
+
+                            if (fut.nodeId().equals(nodeId)) {
+                                SyncMessageAckFuture fut0 = syncMsgFuts.remove(e.getKey());
+
+                                if (fut0 != null) {
+                                    ClusterTopologyException err = new ClusterTopologyException(
+                                        "Node left grid while sending message to: " + nodeId);
+
+                                    fut0.onDone(err);
+                                }
+                            }
+                        }
+
                         break;
 
                     default:
@@ -255,6 +273,11 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
 
                         break;
 
+                    case MSG_EVT_ACK:
+                        processMessageAck(msg);
+
+                        break;
+
                     default:
                         assert false : "Unexpected message received: " + msg.type();
                 }
@@ -343,8 +366,11 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
      * @return Future.
      */
     @SuppressWarnings("TooBroadScope")
-    public IgniteFuture<UUID> startRoutine(GridContinuousHandler hnd, int bufSize, long interval,
-        boolean autoUnsubscribe, @Nullable IgnitePredicate<ClusterNode> prjPred) {
+    public IgniteFuture<UUID> startRoutine(GridContinuousHandler hnd,
+        int bufSize,
+        long interval,
+        boolean autoUnsubscribe,
+        @Nullable IgnitePredicate<ClusterNode> prjPred) {
         assert hnd != null;
         assert bufSize > 0;
         assert interval >= 0;
@@ -432,7 +458,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
             // these nodes.
             for (Map.Entry<UUID, Collection<GridContinuousMessage>> e : pending.entrySet()) {
                 if (nodeIds.add(e.getKey()))
-                    e.getValue().add(new GridContinuousMessage(MSG_START_REQ, routineId, reqData));
+                    e.getValue().add(new GridContinuousMessage(MSG_START_REQ, routineId, null, reqData));
             }
 
             // Register routine locally.
@@ -484,7 +510,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
 
             // Send start requests.
             try {
-                GridContinuousMessage req = new GridContinuousMessage(MSG_START_REQ, routineId, reqData);
+                GridContinuousMessage req = new GridContinuousMessage(MSG_START_REQ, routineId, null, reqData);
 
                 sendWithRetries(nodes, req, null);
             }
@@ -590,13 +616,15 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
                 // Register acknowledge timeout (timeout object will be removed when
                 // future is completed).
                 fut.addTimeoutObject(new StopTimeoutObject(ackTimeout, routineId,
-                    new GridContinuousMessage(MSG_STOP_REQ, routineId, null)));
+                    new GridContinuousMessage(MSG_STOP_REQ, routineId, null, null)));
 
                 // Send stop requests.
                 try {
                     for (ClusterNode node : nodes) {
                         try {
-                            sendWithRetries(node.id(), new GridContinuousMessage(MSG_STOP_REQ, routineId, null), null);
+                            sendWithRetries(node.id(),
+                                new GridContinuousMessage(MSG_STOP_REQ, routineId, null, null),
+                                null);
                         }
                         catch (ClusterTopologyException ignored) {
                             U.warn(log, "Failed to send stop request (node left topology): " + node.id());
@@ -624,11 +652,15 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
      * @param nodeId ID of the node that started routine.
      * @param routineId Routine ID.
      * @param obj Notification object.
-     * @param orderedTopic Topic for ordered notifications.
-     *      If {@code null}, non-ordered message will be sent.
+     * @param orderedTopic Topic for ordered notifications. If {@code null}, non-ordered message will be sent.
+     * @param sync If {@code true} then waits for event acknowledgment.
      * @throws IgniteCheckedException In case of error.
      */
-    public void addNotification(UUID nodeId, UUID routineId, @Nullable Object obj, @Nullable Object orderedTopic)
+    public void addNotification(UUID nodeId,
+        UUID routineId,
+        @Nullable Object obj,
+        @Nullable Object orderedTopic,
+        boolean sync)
         throws IgniteCheckedException {
         assert nodeId != null;
         assert routineId != null;
@@ -638,29 +670,55 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
         RemoteRoutineInfo info = rmtInfos.get(routineId);
 
         if (info != null) {
-            Collection<Object> toSnd = info.add(obj);
+            assert info.interval == 0 || !sync;
+
+            if (sync) {
+                SyncMessageAckFuture fut = new SyncMessageAckFuture(ctx, nodeId);
+
+                IgniteUuid futId = IgniteUuid.randomUuid();
+
+                syncMsgFuts.put(futId, fut);
+
+                try {
+                    sendNotification(nodeId, routineId, futId, F.asList(obj), orderedTopic);
+                }
+                catch (IgniteCheckedException e) {
+                    syncMsgFuts.remove(futId);
 
-            if (toSnd != null)
-                sendNotification(nodeId, routineId, toSnd, orderedTopic);
+                    throw e;
+                }
+
+                fut.get(ackTimeout);
+            }
+            else {
+                Collection<Object> toSnd = info.add(obj);
+
+                if (toSnd != null)
+                    sendNotification(nodeId, routineId, null, toSnd, orderedTopic);
+            }
         }
     }
 
     /**
      * @param nodeId Node ID.
      * @param routineId Routine ID.
+     * @param futId Future ID.
      * @param toSnd Notification object to send.
      * @param orderedTopic Topic for ordered notifications.
      *      If {@code null}, non-ordered message will be sent.
      * @throws IgniteCheckedException In case of error.
      */
-    private void sendNotification(UUID nodeId, UUID routineId, Collection<Object> toSnd,
+    private void sendNotification(UUID nodeId,
+        UUID routineId,
+        @Nullable IgniteUuid futId,
+        Collection<Object> toSnd,
         @Nullable Object orderedTopic) throws IgniteCheckedException {
         assert nodeId != null;
         assert routineId != null;
         assert toSnd != null;
         assert !toSnd.isEmpty();
 
-        sendWithRetries(nodeId, new GridContinuousMessage(MSG_EVT_NOTIFICATION, routineId, toSnd), orderedTopic);
+        sendWithRetries(nodeId, new GridContinuousMessage(MSG_EVT_NOTIFICATION, routineId, futId, toSnd), orderedTopic);
     }
 
     /**
@@ -722,7 +780,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
         }
 
         try {
-            sendWithRetries(nodeId, new GridContinuousMessage(MSG_START_ACK, routineId, err), null);
+            sendWithRetries(nodeId, new GridContinuousMessage(MSG_START_ACK, routineId, null, err), null);
         }
         catch (ClusterTopologyException ignored) {
             if (log.isDebugEnabled())
@@ -783,7 +841,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
         unregisterRemote(routineId);
 
         try {
-            sendWithRetries(nodeId, new GridContinuousMessage(MSG_STOP_ACK, routineId, null), null);
+            sendWithRetries(nodeId, new GridContinuousMessage(MSG_STOP_ACK, routineId, null, null), null);
         }
         catch (ClusterTopologyException ignored) {
             if (log.isDebugEnabled())
@@ -815,19 +873,45 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     }
 
     /**
+     * @param msg Message.
+     */
+    private void processMessageAck(GridContinuousMessage msg) {
+        assert msg.futureId() != null;
+
+        SyncMessageAckFuture fut = syncMsgFuts.remove(msg.futureId());
+
+        if (fut != null)
+            fut.onDone();
+    }
+
+    /**
      * @param nodeId Sender ID.
-     * @param ntf Notification.
+     * @param msg Message.
      */
-    private void processNotification(UUID nodeId, GridContinuousMessage ntf) {
+    private void processNotification(UUID nodeId, GridContinuousMessage msg) {
         assert nodeId != null;
-        assert ntf != null;
+        assert msg != null;
 
-        UUID routineId = ntf.routineId();
+        UUID routineId = msg.routineId();
 
-        LocalRoutineInfo routine = locInfos.get(routineId);
+        try {
+            LocalRoutineInfo routine = locInfos.get(routineId);
 
-        if (routine != null)
-            routine.hnd.notifyCallback(nodeId, routineId, (Collection<?>)ntf.data(), ctx);
+            if (routine != null)
+                routine.hnd.notifyCallback(nodeId, routineId, (Collection<?>) msg.data(), ctx);
+        }
+        finally {
+            if (msg.futureId() != null) {
+                try {
+                    sendWithRetries(nodeId,
+                        new GridContinuousMessage(MSG_EVT_ACK, null, msg.futureId(), null),
+                        null);
+                }
+                catch (IgniteCheckedException e) {
+                    log.error("Failed to send event acknowledgment to node: " + nodeId, e);
+                }
+            }
+        }
     }
 
     /**
@@ -871,8 +955,13 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
      * @return Whether listener was actually registered.
      * @throws IgniteCheckedException In case of error.
      */
-    private boolean registerHandler(final UUID nodeId, final UUID routineId, final GridContinuousHandler hnd,
-        int bufSize, final long interval, boolean autoUnsubscribe, boolean loc) throws IgniteCheckedException {
+    private boolean registerHandler(final UUID nodeId,
+        final UUID routineId,
+        final GridContinuousHandler hnd,
+        int bufSize,
+        final long interval,
+        boolean autoUnsubscribe,
+        boolean loc) throws IgniteCheckedException {
         assert nodeId != null;
         assert routineId != null;
         assert hnd != null;
@@ -915,7 +1004,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
 
                             if (toSnd != null) {
                                 try {
-                                    sendNotification(nodeId, routineId, toSnd, hnd.orderedTopic());
+                                    sendNotification(nodeId, routineId, null, toSnd, hnd.orderedTopic());
                                 }
                                 catch (ClusterTopologyException ignored) {
                                     if (log.isDebugEnabled())
@@ -1637,6 +1726,46 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
     }
 
     /**
+     * Future for stop routine.
+     */
+    private static class SyncMessageAckFuture extends GridFutureAdapter<Object> {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** */
+        private UUID nodeId;
+
+        /**
+         * Required by {@link Externalizable}.
+         */
+        public SyncMessageAckFuture() {
+            // No-op.
+        }
+
+        /**
+         * @param ctx Kernal context.
+         * @param nodeId Master node ID.
+         */
+        SyncMessageAckFuture(GridKernalContext ctx, UUID nodeId) {
+            super(ctx);
+
+            this.nodeId = nodeId;
+        }
+
+        /**
+         * @return Master node ID.
+         */
+        UUID nodeId() {
+            return nodeId;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(SyncMessageAckFuture.class, this);
+        }
+    }
+
+    /**
      * Timeout object for stop process.
      */
     private class StopTimeoutObject extends GridTimeoutObjectAdapter {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
index d8dfc97..5d9edc9 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java
@@ -124,13 +124,13 @@ public class GridServiceProcessor extends GridProcessorAdapter {
 
             cfgQry.localCallback(new DeploymentListener());
 
-            cfgQry.execute(ctx.grid().forLocal(), true, false);
+            cfgQry.execute(ctx.grid().forLocal(), true, false, false);
 
             assignQry = (GridCacheContinuousQueryAdapter<Object, Object>)cache.queries().createContinuousQuery();
 
             assignQry.localCallback(new AssignmentListener());
 
-            assignQry.execute(ctx.grid().forLocal(), true, false);
+            assignQry.execute(ctx.grid().forLocal(), true, false, false);
         }
         finally {
             if (ctx.deploy().enabled())

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
index 6e3221a..04867d9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
@@ -10,6 +10,7 @@
 package org.apache.ignite.internal.processors.cache;
 
 import org.apache.ignite.*;
+import org.apache.ignite.lang.*;
 import org.gridgain.grid.cache.*;
 import org.gridgain.grid.util.typedef.internal.*;
 import org.gridgain.testframework.*;
@@ -17,12 +18,16 @@ import org.jetbrains.annotations.*;
 
 import javax.cache.configuration.*;
 import javax.cache.event.*;
+import javax.cache.expiry.*;
 
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
 
+import static java.util.concurrent.TimeUnit.*;
 import static javax.cache.event.EventType.*;
 import static org.gridgain.grid.cache.GridCacheMode.*;
+import static org.gridgain.grid.cache.GridCachePreloadMode.*;
 
 /**
  *
@@ -35,6 +40,9 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
     private static volatile CountDownLatch evtsLatch;
 
     /** */
+    private static volatile CountDownLatch syncEvtLatch;
+
+    /** */
     private Integer lastKey = 0;
 
     /** */
@@ -48,12 +56,212 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         if (lsnrCfg != null)
             cfg.addCacheEntryListenerConfiguration(lsnrCfg);
 
+        cfg.setEagerTtl(eagerTtl());
+
+        cfg.setPreloadMode(SYNC);
+
         return cfg;
     }
 
     /**
      * @throws Exception If failed.
      */
+    public void testSynchronousEvents() throws Exception {
+        final CacheEntryCreatedListener<Integer, Integer> lsnr = new CreateUpdateRemoveExpireListener() {
+            @Override public void onRemoved(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+                super.onRemoved(evts);
+
+                awaitLatch();
+            }
+
+            @Override public void onCreated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+                super.onCreated(evts);
+
+                awaitLatch();
+            }
+
+            @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+                super.onUpdated(evts);
+
+                awaitLatch();
+            }
+
+            private void awaitLatch() {
+                try {
+                    assertTrue(syncEvtLatch.await(5000, MILLISECONDS));
+                }
+                catch (InterruptedException e) {
+                    fail("Unexpected exception: " + e);
+                }
+            }
+        };
+
+        CacheEntryListenerConfiguration<Integer, Integer> lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+            new Factory<CacheEntryListener<Integer, Integer>>() {
+                @Override public CacheEntryListener<Integer, Integer> create() {
+                    return lsnr;
+                }
+            },
+            null,
+            true,
+            true
+        );
+
+        IgniteCache<Integer, Integer> cache = jcache();
+
+        cache.registerCacheEntryListener(lsnrCfg);
+
+        try {
+            for (Integer key : keys()) {
+                log.info("Check synchronous create event [key=" + key + ']');
+
+                syncEvent(key, 1, cache, 1);
+
+                checkEvent(evts.iterator(), key, CREATED, 1, null);
+
+                log.info("Check synchronous update event [key=" + key + ']');
+
+                syncEvent(key, 2, cache, 1);
+
+                checkEvent(evts.iterator(), key, UPDATED, 2, 1);
+
+                log.info("Check synchronous remove event [key=" + key + ']');
+
+                syncEvent(key, null, cache, 1);
+
+                checkEvent(evts.iterator(), key, REMOVED, null, 2);
+
+                log.info("Check synchronous expire event [key=" + key + ']');
+
+                syncEvent(key,
+                    3,
+                    cache.withExpiryPolicy(new ModifiedExpiryPolicy(new Duration(MILLISECONDS, 1000))),
+                    eagerTtl() ? 2 : 1);
+
+                checkEvent(evts.iterator(), key, CREATED, 3, null);
+
+                if (!eagerTtl()) {
+                    U.sleep(1100);
+
+                    assertNull(primaryCache(key, cache.getName()).get(key));
+
+                    evtsLatch.await(5000, MILLISECONDS);
+
+                    assertEquals(1, evts.size());
+                }
+
+                checkEvent(evts.iterator(), key, EXPIRED, null, 3);
+
+                assertEquals(0, evts.size());
+            }
+        }
+        finally {
+            cache.deregisterCacheEntryListener(lsnrCfg);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testSynchronousEventsListenerNodeFailed() throws Exception {
+        if (cacheMode() != PARTITIONED)
+            return;
+
+        lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+            new Factory<CacheEntryListener<Integer, Integer>>() {
+                @Override public CacheEntryListener<Integer, Integer> create() {
+                    return new NoOpCreateUpdateListener();
+                }
+            },
+            null,
+            true,
+            true
+        );
+
+        final Ignite grid = startGrid(gridCount());
+
+        try {
+            awaitPartitionMapExchange();
+
+            IgniteCache<Integer, Integer> cache = jcache(0);
+
+            Map<Integer, Integer> vals = new HashMap<>();
+
+            for (Integer key : nearKeys(grid.cache(null), 100, 1_000_000))
+                vals.put(key, 1);
+
+            final AtomicBoolean done = new AtomicBoolean();
+
+            IgniteFuture<?> fut = GridTestUtils.runAsync(new Callable<Void>() {
+                @Override public Void call() throws Exception {
+                    U.sleep(500);
+
+                    stopGrid(grid.name());
+
+                    done.set(true);
+
+                    return null;
+                }
+            });
+
+            while (!done.get())
+                cache.putAll(vals);
+
+            fut.get();
+        }
+        finally {
+            stopGrid(gridCount());
+        }
+    }
+
+    /**
+     * @param key Key.
+     * @param val Value.
+     * @param cache Cache.
+     * @param expEvts Expected events number.
+     * @throws Exception If failed.
+     */
+    private void syncEvent(Integer key, Integer val, IgniteCache<Integer, Integer> cache, int expEvts)
+        throws Exception {
+        evts = new ArrayList<>();
+
+        evtsLatch = new CountDownLatch(expEvts);
+
+        syncEvtLatch = new CountDownLatch(1);
+
+        final AtomicBoolean done = new AtomicBoolean();
+
+        IgniteFuture<?> fut = GridTestUtils.runAsync(new Callable<Void>() {
+            @Override public Void call() throws Exception {
+                assertFalse(done.get());
+
+                U.sleep(500);
+
+                assertFalse(done.get());
+
+                syncEvtLatch.countDown();
+
+                return null;
+            }
+        });
+
+        if (val != null)
+            cache.put(key, val);
+        else
+            cache.remove(key);
+
+        done.set(true);
+
+        fut.get();
+
+        evtsLatch.await(5000, MILLISECONDS);
+
+        assertEquals(expEvts, evts.size());
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
     public void testEvents() throws Exception {
         CacheEntryCreatedListener<Integer, Integer> createLsnr = new CacheEntryCreatedListener<Integer, Integer>() {
             @Override public void onCreated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
@@ -76,6 +284,13 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
             }
         };
 
+        CacheEntryExpiredListener<Integer, Integer> expireLsnr = new CacheEntryExpiredListener<Integer, Integer>() {
+            @Override public void onExpired(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+                for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+                    onEvent(evt);
+            }
+        };
+
         IgniteCache<Integer, Integer> cache = jcache();
 
         Map<Integer, Integer> vals = new HashMap<>();
@@ -88,29 +303,33 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         for (Integer key : keys()) {
             log.info("Check create event [key=" + key + ']');
 
-            checkEvents(cache, createLsnr, key, true, false, false);
+            checkEvents(cache, createLsnr, key, true, false, false, false);
 
             log.info("Check update event [key=" + key + ']');
 
-            checkEvents(cache, updateLsnr, key, false, true, false);
+            checkEvents(cache, updateLsnr, key, false, true, false, false);
 
             log.info("Check remove event [key=" + key + ']');
 
-            checkEvents(cache, rmvLsnr, key, false, false, true);
+            checkEvents(cache, rmvLsnr, key, false, false, true, false);
+
+            log.info("Check expire event [key=" + key + ']');
+
+            checkEvents(cache, expireLsnr, key, false, false, false, true);
 
             log.info("Check create/update events [key=" + key + ']');
 
-            checkEvents(cache, new CreateUpdateListener(), key, true, true, false);
+            checkEvents(cache, new CreateUpdateListener(), key, true, true, false, false);
 
-            log.info("Check create/update/remove events [key=" + key + ']');
+            log.info("Check create/update/remove/expire events [key=" + key + ']');
 
-            checkEvents(cache, new CreateUpdateRemoveListener(), key, true, true, true);
+            checkEvents(cache, new CreateUpdateRemoveExpireListener(), key, true, true, true, true);
         }
 
         CacheEntryListenerConfiguration<Integer, Integer> lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
             new Factory<CacheEntryListener<Integer, Integer>>() {
                 @Override public CacheEntryListener<Integer, Integer> create() {
-                    return new CreateUpdateRemoveListener();
+                    return new CreateUpdateRemoveExpireListener();
                 }
             },
             new TestFilterFactory(),
@@ -122,13 +341,21 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
 
         log.info("Check filter.");
 
-        checkFilter(cache, vals);
-
-        cache.deregisterCacheEntryListener(lsnrCfg);
+        try {
+            checkFilter(cache, vals);
+        }
+        finally {
+            cache.deregisterCacheEntryListener(lsnrCfg);
+        }
 
         cache.putAll(vals);
 
-        checkListenerOnStart(vals);
+        try {
+            checkListenerOnStart(vals);
+        }
+        finally {
+            cache.removeAll(vals.keySet());
+        }
     }
 
     /**
@@ -140,7 +367,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
             new Factory<CacheEntryListener<Integer, Integer>>() {
                 @Override public CacheEntryListener<Integer, Integer> create() {
-                    return new CreateUpdateRemoveListener();
+                    return new CreateUpdateRemoveExpireListener();
                 }
             },
             null,
@@ -150,20 +377,25 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
 
         Ignite grid = startGrid(gridCount());
 
-        IgniteCache<Integer, Integer> cache = grid.jcache(null);
+        try {
+            awaitPartitionMapExchange();
 
-        Integer key = Integer.MAX_VALUE;
+            IgniteCache<Integer, Integer> cache = grid.jcache(null);
 
-        log.info("Check create/update/remove events for listener in configuration [key=" + key + ']');
+            Integer key = Integer.MAX_VALUE;
 
-        checkEvents(cache, lsnrCfg, key, true, true, true);
+            log.info("Check create/update/remove events for listener in configuration [key=" + key + ']');
 
-        stopGrid(gridCount());
+            checkEvents(cache, lsnrCfg, key, true, true, true, true);
+        }
+        finally {
+            stopGrid(gridCount());
+        }
 
         lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
             new Factory<CacheEntryListener<Integer, Integer>>() {
                 @Override public CacheEntryListener<Integer, Integer> create() {
-                    return new CreateUpdateRemoveListener();
+                    return new CreateUpdateRemoveExpireListener();
                 }
             },
             new TestFilterFactory(),
@@ -173,13 +405,21 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
 
         grid = startGrid(gridCount());
 
-        cache = grid.jcache(null);
+        try {
+            awaitPartitionMapExchange();
+
+            IgniteCache<Integer, Integer> cache = grid.jcache(null);
 
-        log.info("Check filter for listener in configuration.");
+            log.info("Check filter for listener in configuration.");
 
-        checkFilter(cache, vals);
+            if (cacheMode() == LOCAL)
+                cache.putAll(vals);
 
-        stopGrid(gridCount());
+            checkFilter(cache, vals);
+        }
+        finally {
+            stopGrid(gridCount());
+        }
     }
 
     /**
@@ -189,6 +429,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
      * @param create {@code True} if listens for create events.
      * @param update {@code True} if listens for update events.
      * @param rmv {@code True} if listens for remove events.
+     * @param expire {@code True} if listens for expire events.
      * @throws Exception If failed.
      */
     private void checkEvents(
@@ -197,7 +438,8 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         Integer key,
         boolean create,
         boolean update,
-        boolean rmv) throws Exception {
+        boolean rmv,
+        boolean expire) throws Exception {
         CacheEntryListenerConfiguration<Integer, Integer> lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
             new Factory<CacheEntryListener<Integer, Integer>>() {
                 @Override public CacheEntryListener<Integer, Integer> create() {
@@ -211,7 +453,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
 
         cache.registerCacheEntryListener(lsnrCfg);
 
-        checkEvents(cache, lsnrCfg, key, create, update, rmv);
+        checkEvents(cache, lsnrCfg, key, create, update, rmv, expire);
     }
 
     /**
@@ -222,7 +464,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
     private void checkFilter(IgniteCache<Integer, Integer> cache, Map<Integer, Integer> vals) throws Exception {
         evts = new ArrayList<>();
 
-        final int expEvts = (vals.size() / 2) * 3; // Remove, create and update for half of modified entries.
+        final int expEvts = (vals.size() / 2) * 4; // Remove, create, update and expire for half of modified entries.
 
         evtsLatch = new CountDownLatch(expEvts);
 
@@ -235,55 +477,87 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         for (Integer key : vals.keySet())
             newVals.put(key, -1);
 
-        cache.putAll(newVals);
+        cache.withExpiryPolicy(new ModifiedExpiryPolicy(new Duration(MILLISECONDS, 100))).putAll(newVals);
+
+        U.sleep(200);
 
-        evtsLatch.await(5000, TimeUnit.MILLISECONDS);
+        if (!eagerTtl()) { // Provoke expire events if eager ttl is disabled.
+            for (Integer key : newVals.keySet())
+                assertNull(primaryCache(key, cache.getName()).get(key));
+        }
+
+        evtsLatch.await(5000, MILLISECONDS);
 
         assertEquals(expEvts, evts.size());
 
-        Iterator<CacheEntryEvent<? extends Integer, ? extends Integer>> iter = evts.iterator();
+        Set<Integer> rmvd = new HashSet<>();
+        Set<Integer> created = new HashSet<>();
+        Set<Integer> updated = new HashSet<>();
+        Set<Integer> expired = new HashSet<>();
 
-        for (Integer key : vals.keySet()) {
-            if (key % 2 == 0) {
-                CacheEntryEvent<? extends Integer, ? extends Integer> evt = iter.next();
+        for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts) {
+            assertTrue(evt.getKey() % 2 == 0);
 
-                assertTrue(evt.getKey() % 2 == 0);
-                assertTrue(vals.keySet().contains(evt.getKey()));
-                assertEquals(REMOVED, evt.getEventType());
-                assertNull(evt.getValue());
-                assertEquals(vals.get(evt.getKey()), evt.getOldValue());
+            assertTrue(vals.keySet().contains(evt.getKey()));
 
-                iter.remove();
-            }
-        }
+            switch (evt.getEventType()) {
+                case REMOVED:
+                    assertNull(evt.getValue());
 
-        for (Integer key : vals.keySet()) {
-            if (key % 2 == 0) {
-                CacheEntryEvent<? extends Integer, ? extends Integer> evt = iter.next();
+                    assertEquals(vals.get(evt.getKey()), evt.getOldValue());
 
-                assertTrue(evt.getKey() % 2 == 0);
-                assertTrue(vals.keySet().contains(evt.getKey()));
-                assertEquals(CREATED, evt.getEventType());
-                assertEquals(vals.get(evt.getKey()), evt.getValue());
-                assertNull(evt.getOldValue());
+                    assertTrue(rmvd.add(evt.getKey()));
 
-                iter.remove();
-            }
-        }
+                    break;
+
+                case CREATED:
+                    assertEquals(vals.get(evt.getKey()), evt.getValue());
+
+                    assertNull(evt.getOldValue());
+
+                    assertTrue(rmvd.contains(evt.getKey()));
+
+                    assertTrue(created.add(evt.getKey()));
+
+                    break;
+
+                case UPDATED:
+                    assertEquals(-1, (int)evt.getValue());
+
+                    assertEquals(vals.get(evt.getKey()), evt.getOldValue());
+
+                    assertTrue(rmvd.contains(evt.getKey()));
 
-        for (Integer key : vals.keySet()) {
-            if (key % 2 == 0) {
-                CacheEntryEvent<? extends Integer, ? extends Integer> evt = iter.next();
+                    assertTrue(created.contains(evt.getKey()));
 
-                assertTrue(evt.getKey() % 2 == 0);
-                assertTrue(vals.keySet().contains(evt.getKey()));
-                assertEquals(UPDATED, evt.getEventType());
-                assertEquals(-1, (int) evt.getValue());
-                assertEquals(vals.get(evt.getKey()), evt.getOldValue());
+                    assertTrue(updated.add(evt.getKey()));
 
-                iter.remove();
+                    break;
+
+                case EXPIRED:
+                    assertNull(evt.getValue());
+
+                    assertEquals(-1, (int)evt.getOldValue());
+
+                    assertTrue(rmvd.contains(evt.getKey()));
+
+                    assertTrue(created.contains(evt.getKey()));
+
+                    assertTrue(updated.contains(evt.getKey()));
+
+                    assertTrue(expired.add(evt.getKey()));
+
+                    break;
+
+                default:
+                    fail("Unexpected type: " + evt.getEventType());
             }
         }
+
+        assertEquals(vals.size() / 2, rmvd.size());
+        assertEquals(vals.size() / 2, created.size());
+        assertEquals(vals.size() / 2, updated.size());
+        assertEquals(vals.size() / 2, expired.size());
     }
 
     /**
@@ -293,6 +567,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
      * @param create {@code True} if listens for create events.
      * @param update {@code True} if listens for update events.
      * @param rmv {@code True} if listens for remove events.
+     * @param expire {@code True} if listens for expire events.
      * @throws Exception If failed.
      */
     private void checkEvents(
@@ -301,7 +576,8 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         Integer key,
         boolean create,
         boolean update,
-        boolean rmv) throws Exception {
+        boolean rmv,
+        boolean expire) throws Exception {
         GridTestUtils.assertThrows(log, new Callable<Void>() {
             @Override public Void call() throws Exception {
                 cache.registerCacheEntryListener(lsnrCfg);
@@ -315,7 +591,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         int expEvts = 0;
 
         if (create)
-            expEvts += 2;
+            expEvts += 4;
 
         if (update)
             expEvts += (UPDATES + 1);
@@ -323,6 +599,9 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         if (rmv)
             expEvts += 2;
 
+        if (expire)
+            expEvts += 2;
+
         evts = new ArrayList<>();
 
         evtsLatch = new CountDownLatch(expEvts);
@@ -338,6 +617,16 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
 
         assertTrue(cache.remove(key));
 
+        IgniteCache<Integer, Integer> expirePlcCache =
+            cache.withExpiryPolicy(new CreatedExpiryPolicy(new Duration(MILLISECONDS, 100)));
+
+        expirePlcCache.put(key, 10);
+
+        U.sleep(200);
+
+        if (!eagerTtl())
+            assertNull(primaryCache(key, cache.getName()).get(key)); // Provoke expire event if eager ttl is disabled.
+
         IgniteCache<Integer, Integer> cache1 = cache;
 
         if (gridCount() > 1)
@@ -349,7 +638,17 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
 
         assertTrue(cache1.remove(key));
 
-        evtsLatch.await(5000, TimeUnit.MILLISECONDS);
+        IgniteCache<Integer, Integer> expirePlcCache1 =
+            cache1.withExpiryPolicy(new CreatedExpiryPolicy(new Duration(MILLISECONDS, 100)));
+
+        expirePlcCache1.put(key, 20);
+
+        U.sleep(200);
+
+        if (!eagerTtl())
+            assertNull(primaryCache(key, cache.getName()).get(key)); // Provoke expire event if eager ttl is disabled.
+
+        evtsLatch.await(5000, MILLISECONDS);
 
         assertEquals(expEvts, evts.size());
 
@@ -367,6 +666,12 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
             checkEvent(iter, key, REMOVED, null, UPDATES);
 
         if (create)
+            checkEvent(iter, key, CREATED, 10, null);
+
+        if (expire)
+            checkEvent(iter, key, EXPIRED, null, 10);
+
+        if (create)
             checkEvent(iter, key, CREATED, 1, null);
 
         if (update)
@@ -375,9 +680,15 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         if (rmv)
             checkEvent(iter, key, REMOVED, null, 2);
 
+        if (create)
+            checkEvent(iter, key, CREATED, 20, null);
+
+        if (expire)
+            checkEvent(iter, key, EXPIRED, null, 20);
+
         assertEquals(0, evts.size());
 
-        log.info("Remove listener. ");
+        log.info("Remove listener.");
 
         cache.deregisterCacheEntryListener(lsnrCfg);
 
@@ -453,6 +764,13 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
         return keys;
     }
 
+    /**
+     * @return Value for configuration property {@link GridCacheConfiguration#isEagerTtl()}.
+     */
+    protected boolean eagerTtl() {
+        return true;
+    }
+
     /** {@inheritDoc} */
     @Override protected void afterTestsStopped() throws Exception {
         super.afterTestsStopped();
@@ -466,7 +784,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
      * @param evt Event.
      */
     private static void onEvent(CacheEntryEvent<? extends Integer, ? extends Integer> evt) {
-        //System.out.println("Received event [evt=" + evt + ", thread=" + Thread.currentThread().getName() + ']');
+        // System.out.println("Received event [evt=" + evt + ", thread=" + Thread.currentThread().getName() + ']');
 
         assert evt != null;
         assert evt.getSource() != null : evt;
@@ -524,12 +842,44 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
     /**
      *
      */
-    static class CreateUpdateRemoveListener extends CreateUpdateListener
-        implements CacheEntryRemovedListener<Integer, Integer> {
+    static class NoOpCreateUpdateListener implements CacheEntryCreatedListener<Integer, Integer>,
+        CacheEntryUpdatedListener<Integer, Integer> {
+        /** {@inheritDoc} */
+        @Override public void onCreated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+            for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts) {
+                assert evt != null;
+                assert evt.getSource() != null : evt;
+                assert evt.getEventType() != null : evt;
+                assert evt.getKey() != null : evt;
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+            for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts) {
+                assert evt != null;
+                assert evt.getSource() != null : evt;
+                assert evt.getEventType() != null : evt;
+                assert evt.getKey() != null : evt;
+            }
+        }
+    }
+
+    /**
+     *
+     */
+    static class CreateUpdateRemoveExpireListener extends CreateUpdateListener
+        implements CacheEntryRemovedListener<Integer, Integer>, CacheEntryExpiredListener<Integer, Integer> {
         /** {@inheritDoc} */
         @Override public void onRemoved(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
             for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
                 onEvent(evt);
         }
+
+        /** {@inheritDoc} */
+        @Override public void onExpired(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> evts) {
+            for (CacheEntryEvent<? extends Integer, ? extends Integer> evt : evts)
+                onEvent(evt);
+        }
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerEagerTtlDisabledTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerEagerTtlDisabledTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerEagerTtlDisabledTest.java
new file mode 100644
index 0000000..6b7110d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerEagerTtlDisabledTest.java
@@ -0,0 +1,22 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.gridgain.grid.cache.*;
+
+/**
+ * Tests expire events when {@link GridCacheConfiguration#isEagerTtl()} is disabled.
+ */
+public class IgniteCacheEntryListenerEagerTtlDisabledTest extends IgniteCacheEntryListenerTxTest {
+    /** {@inheritDoc} */
+    @Override protected boolean eagerTtl() {
+        return false;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/test/java/org/gridgain/testframework/junits/common/GridCommonAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/gridgain/testframework/junits/common/GridCommonAbstractTest.java
index f40d941..25f954a 100644
--- a/modules/core/src/test/java/org/gridgain/testframework/junits/common/GridCommonAbstractTest.java
+++ b/modules/core/src/test/java/org/gridgain/testframework/junits/common/GridCommonAbstractTest.java
@@ -480,6 +480,19 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
     }
 
     /**
+     * @param key Key.
+     * @param cacheName Cache name.
+     * @return Cache.
+     */
+    protected <K, V> IgniteCache<K, V> primaryCache(Object key, @Nullable String cacheName) {
+        ClusterNode node = grid(0).cache(cacheName).affinity().mapKeyToNode(key);
+
+        assertNotNull(node);
+
+        return grid((String)node.attribute(GridNodeAttributes.ATTR_GRID_NAME)).jcache(cacheName);
+    }
+
+    /**
      * @param cache Cache.
      * @return Collection of keys for which given cache is primary.
      * @throws IgniteCheckedException If failed.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/13740b70/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java b/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
index efe49fb..6c01b2f 100644
--- a/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
+++ b/modules/core/src/test/java/org/gridgain/testsuites/bamboo/GridDataGridTestSuite.java
@@ -44,6 +44,7 @@ public class GridDataGridTestSuite extends TestSuite {
         suite.addTestSuite(IgniteCacheEntryListenerTxTest.class);
         suite.addTestSuite(IgniteCacheEntryListenerTxReplicatedTest.class);
         suite.addTestSuite(IgniteCacheEntryListenerTxLocalTest.class);
+        suite.addTestSuite(IgniteCacheEntryListenerEagerTtlDisabledTest.class);
 
         suite.addTest(IgniteCacheExpiryPolicyTestSuite.suite());
 


Mime
View raw message