# sprint-1 moved classes to internal packages
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/ac33a7f0 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/ac33a7f0 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/ac33a7f0 Branch: refs/heads/ignite-sql-tests Commit: ac33a7f03a0f831f442bf946e6a6b01d9987b251 Parents: d58414c Author: sboikov <sboi...@gridgain.com> Authored: Thu Feb 5 13:19:57 2015 +0300 Committer: sboikov <sboi...@gridgain.com> Committed: Thu Feb 5 13:19:57 2015 +0300 ---------------------------------------------------------------------- .../apache/ignite/cache/CacheEntryEvent.java | 75 ----- .../cache/store/CacheStoreBalancingWrapper.java | 296 ------------------ .../processors/cache/CacheEntryEvent.java | 75 +++++ .../cache/CacheStoreBalancingWrapper.java | 297 +++++++++++++++++++ .../GridCacheContinuousQueryManager.java | 19 +- .../store/GridCacheBalancingStoreSelfTest.java | 1 + .../GridCacheOffHeapTieredAbstractSelfTest.java | 4 +- 7 files changed, 385 insertions(+), 382 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ac33a7f0/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java b/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java deleted file mode 100644 index 904bf35..0000000 --- a/modules/core/src/main/java/org/apache/ignite/cache/CacheEntryEvent.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.cache; - -import org.apache.ignite.*; -import org.apache.ignite.cache.query.*; - -import javax.cache.event.*; - -/** - * Implementation of {@link org.apache.ignite.cache.CacheEntryEvent}. - */ -public class CacheEntryEvent<K, V> extends javax.cache.event.CacheEntryEvent<K, V> { - /** */ - private final CacheContinuousQueryEntry<K, V> e; - - /** - * @param src Cache. - * @param type Event type. - * @param e Ignite event. - */ - public CacheEntryEvent(IgniteCache src, EventType type, CacheContinuousQueryEntry<K, V> e) { - super(src, type); - - this.e = e; - } - - /** {@inheritDoc} */ - @Override public V getOldValue() { - return e.getOldValue(); - } - - /** {@inheritDoc} */ - @Override public boolean isOldValueAvailable() { - return e.getOldValue() != null; - } - - /** {@inheritDoc} */ - @Override public K getKey() { - return e.getKey(); - } - - /** {@inheritDoc} */ - @Override public V getValue() { - return e.getValue(); - } - - /** {@inheritDoc} */ - @Override public <T> T unwrap(Class<T> cls) { - throw new IllegalArgumentException(); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return "CacheEntryEvent [evtType=" + getEventType() + - ", key=" + getKey() + - ", val=" + getValue() + - ", oldVal=" + getOldValue() + ']'; - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ac33a7f0/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreBalancingWrapper.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreBalancingWrapper.java b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreBalancingWrapper.java deleted file mode 100644 index 516cd3f..0000000 --- a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreBalancingWrapper.java +++ /dev/null @@ -1,296 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.cache.store; - -import org.apache.ignite.*; -import org.apache.ignite.internal.util.future.*; -import org.apache.ignite.internal.util.typedef.*; -import org.apache.ignite.lang.*; -import org.jdk8.backport.*; -import org.jetbrains.annotations.*; - -import javax.cache.*; -import javax.cache.integration.*; -import java.util.*; -import java.util.concurrent.*; - -/** - * Cache store wrapper that ensures that there will be no more that one thread loading value from underlying store. - */ -public class CacheStoreBalancingWrapper<K, V> extends CacheStore<K, V> { - /** */ - public static final int DFLT_LOAD_ALL_THRESHOLD = 5; - - /** Delegate store. */ - private CacheStore<K, V> delegate; - - /** Pending cache store loads. */ - private ConcurrentMap<K, LoadFuture> pendingLoads = new ConcurrentHashMap8<>(); - - /** Load all threshold. */ - private int loadAllThreshold = DFLT_LOAD_ALL_THRESHOLD; - - /** - * @param delegate Delegate store. - */ - public CacheStoreBalancingWrapper(CacheStore<K, V> delegate) { - this.delegate = delegate; - } - - /** - * @param delegate Delegate store. - * @param loadAllThreshold Load all threshold. - */ - public CacheStoreBalancingWrapper(CacheStore<K, V> delegate, int loadAllThreshold) { - this.delegate = delegate; - this.loadAllThreshold = loadAllThreshold; - } - - /** - * @return Load all threshold. - */ - public int loadAllThreshold() { - return loadAllThreshold; - } - - /** {@inheritDoc} */ - @Nullable @Override public V load(K key) { - LoadFuture fut = pendingLoads.get(key); - - try { - if (fut != null) - return fut.get(key); - - fut = new LoadFuture(); - - LoadFuture old = pendingLoads.putIfAbsent(key, fut); - - if (old != null) - return old.get(key); - } - catch (IgniteCheckedException e) { - throw new CacheLoaderException(e); - } - - try { - V val = delegate.load(key); - - fut.onComplete(key, val); - - return val; - } - catch (Throwable e) { - fut.onError(key, e); - - throw e; - } - } - - /** {@inheritDoc} */ - @Override public void loadCache(IgniteBiInClosure<K, V> clo, @Nullable Object... args) { - delegate.loadCache(clo, args); - } - - /** {@inheritDoc} */ - @Override public Map<K, V> loadAll(Iterable<? extends K> keys) throws CacheLoaderException { - assert false; - - return delegate.loadAll(keys); - } - - /** - * @param keys Keys to load. - * @param c Closure for loaded values. - */ - public void loadAll(Collection<? extends K> keys, final IgniteBiInClosure<K, V> c) { - assert keys.size() <= loadAllThreshold : loadAllThreshold; - - Collection<K> needLoad = null; - Map<K, LoadFuture> pending = null; - LoadFuture span = null; - - for (K key : keys) { - LoadFuture fut = pendingLoads.get(key); - - if (fut != null) { - if (pending == null) - pending = new HashMap<>(); - - pending.put(key, fut); - } - else { - // Try to concurrently add pending future. - if (span == null) - span = new LoadFuture(); - - LoadFuture old = pendingLoads.putIfAbsent(key, span); - - if (old != null) { - if (pending == null) - pending = new HashMap<>(); - - pending.put(key, old); - } - else { - if (needLoad == null) - needLoad = new ArrayList<>(keys.size()); - - needLoad.add(key); - } - } - } - - if (needLoad != null) { - assert !needLoad.isEmpty(); - assert span != null; - - try { - Map<K, V> loaded = delegate.loadAll(needLoad); - - if (loaded != null) { - for (Map.Entry<K, V> e : loaded.entrySet()) - c.apply(e.getKey(), e.getValue()); - } - - span.onComplete(needLoad, loaded); - } - catch (Throwable e) { - span.onError(needLoad, e); - - throw e; - } - } - - if (pending != null) { - try { - for (Map.Entry<K, LoadFuture> e : pending.entrySet()) { - K key = e.getKey(); - - c.apply(key, e.getValue().get(key)); - } - } - catch (IgniteCheckedException e) { - throw new CacheLoaderException(e); - } - } - } - - /** {@inheritDoc} */ - @Override public void write(Cache.Entry<? extends K, ? extends V> entry) { - delegate.write(entry); - } - - /** {@inheritDoc} */ - @Override public void writeAll(Collection<Cache.Entry<? extends K, ? extends V>> entries) { - delegate.writeAll(entries); - } - - /** {@inheritDoc} */ - @Override public void delete(Object key) throws CacheWriterException { - delegate.delete(key); - } - - /** {@inheritDoc} */ - @Override public void deleteAll(Collection<?> keys) throws CacheWriterException { - delegate.deleteAll(keys); - } - - /** {@inheritDoc} */ - @Override public void txEnd(boolean commit) { - delegate.txEnd(commit); - } - - /** - * - */ - private class LoadFuture extends GridFutureAdapter<Map<K, V>> { - /** */ - private static final long serialVersionUID = 0L; - - /** Collection of keys for pending cleanup. */ - private volatile Collection<K> keys; - - /** - * - */ - public LoadFuture() { - // No-op. - } - - /** {@inheritDoc} */ - @Override public boolean onDone(@Nullable Map<K, V> res, @Nullable Throwable err) { - if (super.onDone(res, err)) { - assert keys != null; - - for (K key : keys) - pendingLoads.remove(key, this); - - return true; - } - - return false; - } - - /** - * @param key Key. - * @param val Loaded value. - */ - public void onComplete(K key, V val) { - onComplete(Collections.singletonList(key), F.asMap(key, val)); - } - - /** - * @param keys Keys. - * @param res Loaded values. - */ - public void onComplete(Collection<K> keys, Map<K, V> res) { - this.keys = keys; - - onDone(res); - } - - /** - * @param key Key. - * @param err Error. - */ - public void onError(K key, Throwable err) { - - } - - /** - * @param keys Keys. - * @param err Error. - */ - public void onError(Collection<K> keys, Throwable err) { - this.keys = keys; - - onDone(err); - } - - /** - * Gets value loaded for key k. - * - * @param key Key to load. - * @return Loaded value (possibly {@code null}). - * @throws IgniteCheckedException If load failed. - */ - public V get(K key) throws IgniteCheckedException { - return get().get(key); - } - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ac33a7f0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryEvent.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryEvent.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryEvent.java new file mode 100644 index 0000000..1ff4be8 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheEntryEvent.java @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cache; + +import org.apache.ignite.*; +import org.apache.ignite.cache.query.*; + +import javax.cache.event.*; + +/** + * Implementation of {@link javax.cache.event.CacheEntryEvent}. + */ +public class CacheEntryEvent<K, V> extends javax.cache.event.CacheEntryEvent<K, V> { + /** */ + private final CacheContinuousQueryEntry<K, V> e; + + /** + * @param src Cache. + * @param type Event type. + * @param e Ignite event. + */ + public CacheEntryEvent(IgniteCache src, EventType type, CacheContinuousQueryEntry<K, V> e) { + super(src, type); + + this.e = e; + } + + /** {@inheritDoc} */ + @Override public V getOldValue() { + return e.getOldValue(); + } + + /** {@inheritDoc} */ + @Override public boolean isOldValueAvailable() { + return e.getOldValue() != null; + } + + /** {@inheritDoc} */ + @Override public K getKey() { + return e.getKey(); + } + + /** {@inheritDoc} */ + @Override public V getValue() { + return e.getValue(); + } + + /** {@inheritDoc} */ + @Override public <T> T unwrap(Class<T> cls) { + throw new IllegalArgumentException(); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return "CacheEntryEvent [evtType=" + getEventType() + + ", key=" + getKey() + + ", val=" + getValue() + + ", oldVal=" + getOldValue() + ']'; + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ac33a7f0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java new file mode 100644 index 0000000..9013fcb --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java @@ -0,0 +1,297 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cache; + +import org.apache.ignite.*; +import org.apache.ignite.cache.store.*; +import org.apache.ignite.internal.util.future.*; +import org.apache.ignite.internal.util.typedef.*; +import org.apache.ignite.lang.*; +import org.jdk8.backport.*; +import org.jetbrains.annotations.*; + +import javax.cache.*; +import javax.cache.integration.*; +import java.util.*; +import java.util.concurrent.*; + +/** + * Cache store wrapper that ensures that there will be no more that one thread loading value from underlying store. + */ +public class CacheStoreBalancingWrapper<K, V> extends CacheStore<K, V> { + /** */ + public static final int DFLT_LOAD_ALL_THRESHOLD = 5; + + /** Delegate store. */ + private CacheStore<K, V> delegate; + + /** Pending cache store loads. */ + private ConcurrentMap<K, LoadFuture> pendingLoads = new ConcurrentHashMap8<>(); + + /** Load all threshold. */ + private int loadAllThreshold = DFLT_LOAD_ALL_THRESHOLD; + + /** + * @param delegate Delegate store. + */ + public CacheStoreBalancingWrapper(CacheStore<K, V> delegate) { + this.delegate = delegate; + } + + /** + * @param delegate Delegate store. + * @param loadAllThreshold Load all threshold. + */ + public CacheStoreBalancingWrapper(CacheStore<K, V> delegate, int loadAllThreshold) { + this.delegate = delegate; + this.loadAllThreshold = loadAllThreshold; + } + + /** + * @return Load all threshold. + */ + public int loadAllThreshold() { + return loadAllThreshold; + } + + /** {@inheritDoc} */ + @Nullable @Override public V load(K key) { + LoadFuture fut = pendingLoads.get(key); + + try { + if (fut != null) + return fut.get(key); + + fut = new LoadFuture(); + + LoadFuture old = pendingLoads.putIfAbsent(key, fut); + + if (old != null) + return old.get(key); + } + catch (IgniteCheckedException e) { + throw new CacheLoaderException(e); + } + + try { + V val = delegate.load(key); + + fut.onComplete(key, val); + + return val; + } + catch (Throwable e) { + fut.onError(key, e); + + throw e; + } + } + + /** {@inheritDoc} */ + @Override public void loadCache(IgniteBiInClosure<K, V> clo, @Nullable Object... args) { + delegate.loadCache(clo, args); + } + + /** {@inheritDoc} */ + @Override public Map<K, V> loadAll(Iterable<? extends K> keys) throws CacheLoaderException { + assert false; + + return delegate.loadAll(keys); + } + + /** + * @param keys Keys to load. + * @param c Closure for loaded values. + */ + public void loadAll(Collection<? extends K> keys, final IgniteBiInClosure<K, V> c) { + assert keys.size() <= loadAllThreshold : loadAllThreshold; + + Collection<K> needLoad = null; + Map<K, LoadFuture> pending = null; + LoadFuture span = null; + + for (K key : keys) { + LoadFuture fut = pendingLoads.get(key); + + if (fut != null) { + if (pending == null) + pending = new HashMap<>(); + + pending.put(key, fut); + } + else { + // Try to concurrently add pending future. + if (span == null) + span = new LoadFuture(); + + LoadFuture old = pendingLoads.putIfAbsent(key, span); + + if (old != null) { + if (pending == null) + pending = new HashMap<>(); + + pending.put(key, old); + } + else { + if (needLoad == null) + needLoad = new ArrayList<>(keys.size()); + + needLoad.add(key); + } + } + } + + if (needLoad != null) { + assert !needLoad.isEmpty(); + assert span != null; + + try { + Map<K, V> loaded = delegate.loadAll(needLoad); + + if (loaded != null) { + for (Map.Entry<K, V> e : loaded.entrySet()) + c.apply(e.getKey(), e.getValue()); + } + + span.onComplete(needLoad, loaded); + } + catch (Throwable e) { + span.onError(needLoad, e); + + throw e; + } + } + + if (pending != null) { + try { + for (Map.Entry<K, LoadFuture> e : pending.entrySet()) { + K key = e.getKey(); + + c.apply(key, e.getValue().get(key)); + } + } + catch (IgniteCheckedException e) { + throw new CacheLoaderException(e); + } + } + } + + /** {@inheritDoc} */ + @Override public void write(Cache.Entry<? extends K, ? extends V> entry) { + delegate.write(entry); + } + + /** {@inheritDoc} */ + @Override public void writeAll(Collection<Cache.Entry<? extends K, ? extends V>> entries) { + delegate.writeAll(entries); + } + + /** {@inheritDoc} */ + @Override public void delete(Object key) throws CacheWriterException { + delegate.delete(key); + } + + /** {@inheritDoc} */ + @Override public void deleteAll(Collection<?> keys) throws CacheWriterException { + delegate.deleteAll(keys); + } + + /** {@inheritDoc} */ + @Override public void txEnd(boolean commit) { + delegate.txEnd(commit); + } + + /** + * + */ + private class LoadFuture extends GridFutureAdapter<Map<K, V>> { + /** */ + private static final long serialVersionUID = 0L; + + /** Collection of keys for pending cleanup. */ + private volatile Collection<K> keys; + + /** + * + */ + public LoadFuture() { + // No-op. + } + + /** {@inheritDoc} */ + @Override public boolean onDone(@Nullable Map<K, V> res, @Nullable Throwable err) { + if (super.onDone(res, err)) { + assert keys != null; + + for (K key : keys) + pendingLoads.remove(key, this); + + return true; + } + + return false; + } + + /** + * @param key Key. + * @param val Loaded value. + */ + public void onComplete(K key, V val) { + onComplete(Collections.singletonList(key), F.asMap(key, val)); + } + + /** + * @param keys Keys. + * @param res Loaded values. + */ + public void onComplete(Collection<K> keys, Map<K, V> res) { + this.keys = keys; + + onDone(res); + } + + /** + * @param key Key. + * @param err Error. + */ + public void onError(K key, Throwable err) { + + } + + /** + * @param keys Keys. + * @param err Error. + */ + public void onError(Collection<K> keys, Throwable err) { + this.keys = keys; + + onDone(err); + } + + /** + * Gets value loaded for key k. + * + * @param key Key to load. + * @return Loaded value (possibly {@code null}). + * @throws IgniteCheckedException If load failed. + */ + public V get(K key) throws IgniteCheckedException { + return get().get(key); + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ac33a7f0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java index fb4c476..bd70f02 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/GridCacheContinuousQueryManager.java @@ -21,6 +21,7 @@ import org.apache.ignite.*; import org.apache.ignite.cache.*; import org.apache.ignite.cache.query.*; import org.apache.ignite.internal.processors.cache.*; +import org.apache.ignite.internal.processors.cache.CacheEntryEvent; import org.apache.ignite.internal.util.typedef.*; import org.apache.ignite.internal.util.typedef.internal.*; import org.apache.ignite.lang.*; @@ -610,7 +611,7 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt assert cache != null : cacheName; } - return fltr.evaluate(new org.apache.ignite.cache.CacheEntryEvent(cache, evtType, entry)); + return fltr.evaluate(new CacheEntryEvent(cache, evtType, entry)); } catch (Exception e) { LT.warn(ignite.log(), e, "Cache entry event filter error: " + e); @@ -727,8 +728,8 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt case EXPIRED: { assert expireLsnr != null; - org.apache.ignite.cache.CacheEntryEvent evt0 = - new org.apache.ignite.cache.CacheEntryEvent(cache, EXPIRED, entry); + CacheEntryEvent evt0 = + new CacheEntryEvent(cache, EXPIRED, entry); expireLsnr.onExpired(Collections.singleton(evt0)); @@ -738,8 +739,8 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt case REMOVED: { assert rmvLsnr != null; - org.apache.ignite.cache.CacheEntryEvent evt0 = - new org.apache.ignite.cache.CacheEntryEvent(cache, REMOVED, entry); + CacheEntryEvent evt0 = + new CacheEntryEvent(cache, REMOVED, entry); rmvLsnr.onRemoved(Collections.singleton(evt0)); @@ -749,8 +750,8 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt case UPDATED: { assert updateLsnr != null; - org.apache.ignite.cache.CacheEntryEvent evt0 = - new org.apache.ignite.cache.CacheEntryEvent(cache, UPDATED, entry); + CacheEntryEvent evt0 = + new CacheEntryEvent(cache, UPDATED, entry); updateLsnr.onUpdated(Collections.singleton(evt0)); @@ -760,8 +761,8 @@ public class GridCacheContinuousQueryManager<K, V> extends GridCacheManagerAdapt case CREATED: { assert createLsnr != null; - org.apache.ignite.cache.CacheEntryEvent evt0 = - new org.apache.ignite.cache.CacheEntryEvent(cache, CREATED, entry); + CacheEntryEvent evt0 = + new CacheEntryEvent(cache, CREATED, entry); createLsnr.onCreated(Collections.singleton(evt0)); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ac33a7f0/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java index a3b1b72..38055dc 100644 --- a/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java @@ -18,6 +18,7 @@ package org.apache.ignite.cache.store; import org.apache.ignite.internal.*; +import org.apache.ignite.internal.processors.cache.*; import org.apache.ignite.internal.util.typedef.*; import org.apache.ignite.lang.*; import org.apache.ignite.testframework.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ac33a7f0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapTieredAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapTieredAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapTieredAbstractSelfTest.java index ab6b5fb..bf607b4 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapTieredAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheOffHeapTieredAbstractSelfTest.java @@ -530,7 +530,7 @@ public abstract class GridCacheOffHeapTieredAbstractSelfTest extends GridCacheAb c.put(key, val); - assertNull(c.localPeek(key)); + assertNull(c.localPeek(key, CachePeekMode.ONHEAP)); Lock lock = c.lock(key); @@ -542,7 +542,7 @@ public abstract class GridCacheOffHeapTieredAbstractSelfTest extends GridCacheAb assertFalse(c.isLocalLocked(key, false)); - assertNull(c.localPeek(key)); + assertNull(c.localPeek(key, CachePeekMode.ONHEAP)); checkValue(key, val); }
