You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2016/01/19 10:12:41 UTC
[01/28] ignite git commit:
IgniteHadoopIgfsSecondaryFileSystem.usrName field is renamed to "userName" to
preserve backward compatibility.
Repository: ignite
Updated Branches:
refs/heads/ignite-2236 7d8e304d9 -> 991eef2a3
IgniteHadoopIgfsSecondaryFileSystem.usrName field is renamed to "userName" to preserve backward compatibility.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/122d27c8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/122d27c8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/122d27c8
Branch: refs/heads/ignite-2236
Commit: 122d27c831903ba00b9409ef21345b4a68ff5ec4
Parents: d881417
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Thu Jan 7 17:48:14 2016 +0400
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 11:35:39 2016 +0300
----------------------------------------------------------------------
.../ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/122d27c8/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
index 9f544c1..12cd2ac 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
@@ -114,14 +114,14 @@ public class IgniteHadoopIgfsSecondaryFileSystem implements IgfsSecondaryFileSys
*
* @param uri URI of file system.
* @param cfgPath Additional path to Hadoop configuration.
- * @param usrName User name.
+ * @param userName User name.
* @throws IgniteCheckedException In case of error.
* @deprecated Use {@link #getFileSystemFactory()} instead.
*/
@Deprecated
public IgniteHadoopIgfsSecondaryFileSystem(@Nullable String uri, @Nullable String cfgPath,
- @Nullable String usrName) throws IgniteCheckedException {
- setDefaultUserName(usrName);
+ @Nullable String userName) throws IgniteCheckedException {
+ setDefaultUserName(userName);
CachingHadoopFileSystemFactory fac = new CachingHadoopFileSystemFactory();
[21/28] ignite git commit: IGNITE-1514: .NET: Added "DestroyCache"
method. This closes #375.
Posted by vo...@apache.org.
IGNITE-1514: .NET: Added "DestroyCache" method. This closes #375.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/9a996332
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/9a996332
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/9a996332
Branch: refs/heads/ignite-2236
Commit: 9a996332df81aee79b629811062d28fe75dead99
Parents: d7fd580
Author: Pavel Tupitsyn <pt...@gridgain.com>
Authored: Mon Jan 18 18:42:34 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 18:42:34 2016 +0300
----------------------------------------------------------------------
.../platform/PlatformNoopProcessor.java | 5 +++++
.../processors/platform/PlatformProcessor.java | 8 ++++++++
.../platform/PlatformProcessorImpl.java | 5 +++++
.../cpp/common/include/ignite/common/exports.h | 1 +
.../cpp/common/include/ignite/common/java.h | 3 +++
.../platforms/cpp/common/project/vs/module.def | 3 ++-
modules/platforms/cpp/common/src/exports.cpp | 4 ++++
modules/platforms/cpp/common/src/java.cpp | 20 ++++++++++++++++++++
.../Cache/CacheAbstractTest.cs | 19 +++++++++++++++++++
.../dotnet/Apache.Ignite.Core/IIgnite.cs | 7 +++++++
.../dotnet/Apache.Ignite.Core/Impl/Ignite.cs | 6 ++++++
.../Apache.Ignite.Core/Impl/IgniteProxy.cs | 6 ++++++
.../Impl/Unmanaged/IgniteJniNativeMethods.cs | 3 +++
.../Impl/Unmanaged/UnmanagedUtils.cs | 16 ++++++++++++++++
14 files changed, 105 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformNoopProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformNoopProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformNoopProcessor.java
index 9142543..fb28008 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformNoopProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformNoopProcessor.java
@@ -74,6 +74,11 @@ public class PlatformNoopProcessor extends GridProcessorAdapter implements Platf
}
/** {@inheritDoc} */
+ @Override public void destroyCache(@Nullable String name) throws IgniteCheckedException {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
@Override public PlatformTarget affinity(@Nullable String name) throws IgniteCheckedException {
return null;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessor.java
index fa22953..8e684e3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessor.java
@@ -90,6 +90,14 @@ public interface PlatformProcessor extends GridProcessor {
public PlatformTarget getOrCreateCache(@Nullable String name) throws IgniteCheckedException;
/**
+ * Destroy dynamically created cache.
+ *
+ * @param name Cache name.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void destroyCache(@Nullable String name) throws IgniteCheckedException;
+
+ /**
* Get affinity.
*
* @param name Cache name.
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
index b0870bb..dc6e0df 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
@@ -244,6 +244,11 @@ public class PlatformProcessorImpl extends GridProcessorAdapter implements Platf
}
/** {@inheritDoc} */
+ @Override public void destroyCache(@Nullable String name) throws IgniteCheckedException {
+ ctx.grid().destroyCache(name);
+ }
+
+ /** {@inheritDoc} */
@Override public PlatformTarget affinity(@Nullable String name) throws IgniteCheckedException {
return new PlatformAffinity(platformCtx, ctx, name);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/cpp/common/include/ignite/common/exports.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/common/exports.h b/modules/platforms/cpp/common/include/ignite/common/exports.h
index 3eb775d..67583ed 100644
--- a/modules/platforms/cpp/common/include/ignite/common/exports.h
+++ b/modules/platforms/cpp/common/include/ignite/common/exports.h
@@ -36,6 +36,7 @@ extern "C" {
void* IGNITE_CALL IgniteProcessorCache(gcj::JniContext* ctx, void* obj, char* name);
void* IGNITE_CALL IgniteProcessorCreateCache(gcj::JniContext* ctx, void* obj, char* name);
void* IGNITE_CALL IgniteProcessorGetOrCreateCache(gcj::JniContext* ctx, void* obj, char* name);
+ void IGNITE_CALL IgniteProcessorDestroyCache(gcj::JniContext* ctx, void* obj, char* name);
void* IGNITE_CALL IgniteProcessorAffinity(gcj::JniContext* ctx, void* obj, char* name);
void* IGNITE_CALL IgniteProcessorDataStreamer(gcj::JniContext* ctx, void* obj, char* name, bool keepPortable);
void* IGNITE_CALL IgniteProcessorTransactions(gcj::JniContext* ctx, void* obj);
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/cpp/common/include/ignite/common/java.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/common/java.h b/modules/platforms/cpp/common/include/ignite/common/java.h
index e629c77..572f040 100644
--- a/modules/platforms/cpp/common/include/ignite/common/java.h
+++ b/modules/platforms/cpp/common/include/ignite/common/java.h
@@ -297,6 +297,7 @@ namespace ignite
jmethodID m_PlatformProcessor_cache;
jmethodID m_PlatformProcessor_createCache;
jmethodID m_PlatformProcessor_getOrCreateCache;
+ jmethodID m_PlatformProcessor_destroyCache;
jmethodID m_PlatformProcessor_affinity;
jmethodID m_PlatformProcessor_dataStreamer;
jmethodID m_PlatformProcessor_transactions;
@@ -488,6 +489,8 @@ namespace ignite
jobject ProcessorCreateCache(jobject obj, const char* name, JniErrorInfo* errInfo);
jobject ProcessorGetOrCreateCache(jobject obj, const char* name);
jobject ProcessorGetOrCreateCache(jobject obj, const char* name, JniErrorInfo* errInfo);
+ void ProcessorDestroyCache(jobject obj, const char* name);
+ void ProcessorDestroyCache(jobject obj, const char* name, JniErrorInfo* errInfo);
jobject ProcessorAffinity(jobject obj, const char* name);
jobject ProcessorDataStreamer(jobject obj, const char* name, bool keepPortable);
jobject ProcessorTransactions(jobject obj);
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/cpp/common/project/vs/module.def
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/project/vs/module.def b/modules/platforms/cpp/common/project/vs/module.def
index 3d166bd..c947128 100644
--- a/modules/platforms/cpp/common/project/vs/module.def
+++ b/modules/platforms/cpp/common/project/vs/module.def
@@ -112,4 +112,5 @@ IgniteAtomicLongClose @109
IgniteListenableCancel @110
IgniteListenableIsCancelled @111
IgniteTargetListenFutureAndGet @112
-IgniteTargetListenFutureForOperationAndGet @113
\ No newline at end of file
+IgniteTargetListenFutureForOperationAndGet @113
+IgniteProcessorDestroyCache @114
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/cpp/common/src/exports.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/src/exports.cpp b/modules/platforms/cpp/common/src/exports.cpp
index 08425a4..d4ffa7e 100644
--- a/modules/platforms/cpp/common/src/exports.cpp
+++ b/modules/platforms/cpp/common/src/exports.cpp
@@ -66,6 +66,10 @@ extern "C" {
return ctx->ProcessorGetOrCreateCache(static_cast<jobject>(obj), name);
}
+ void IGNITE_CALL IgniteProcessorDestroyCache(gcj::JniContext* ctx, void* obj, char* name) {
+ ctx->ProcessorDestroyCache(static_cast<jobject>(obj), name);
+ }
+
void* IGNITE_CALL IgniteProcessorAffinity(gcj::JniContext* ctx, void* obj, char* name) {
return ctx->ProcessorAffinity(static_cast<jobject>(obj), name);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/cpp/common/src/java.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/src/java.cpp b/modules/platforms/cpp/common/src/java.cpp
index 63deba5..9e55742 100644
--- a/modules/platforms/cpp/common/src/java.cpp
+++ b/modules/platforms/cpp/common/src/java.cpp
@@ -191,6 +191,7 @@ namespace ignite
JniMethod M_PLATFORM_PROCESSOR_CACHE = JniMethod("cache", "(Ljava/lang/String;)Lorg/apache/ignite/internal/processors/platform/PlatformTarget;", false);
JniMethod M_PLATFORM_PROCESSOR_CREATE_CACHE = JniMethod("createCache", "(Ljava/lang/String;)Lorg/apache/ignite/internal/processors/platform/PlatformTarget;", false);
JniMethod M_PLATFORM_PROCESSOR_GET_OR_CREATE_CACHE = JniMethod("getOrCreateCache", "(Ljava/lang/String;)Lorg/apache/ignite/internal/processors/platform/PlatformTarget;", false);
+ JniMethod M_PLATFORM_PROCESSOR_DESTROY_CACHE = JniMethod("destroyCache", "(Ljava/lang/String;)V", false);
JniMethod M_PLATFORM_PROCESSOR_AFFINITY = JniMethod("affinity", "(Ljava/lang/String;)Lorg/apache/ignite/internal/processors/platform/PlatformTarget;", false);
JniMethod M_PLATFORM_PROCESSOR_DATA_STREAMER = JniMethod("dataStreamer", "(Ljava/lang/String;Z)Lorg/apache/ignite/internal/processors/platform/PlatformTarget;", false);
JniMethod M_PLATFORM_PROCESSOR_TRANSACTIONS = JniMethod("transactions", "()Lorg/apache/ignite/internal/processors/platform/PlatformTarget;", false);
@@ -635,6 +636,7 @@ namespace ignite
m_PlatformProcessor_cache = FindMethod(env, c_PlatformProcessor, M_PLATFORM_PROCESSOR_CACHE);
m_PlatformProcessor_createCache = FindMethod(env, c_PlatformProcessor, M_PLATFORM_PROCESSOR_CREATE_CACHE);
m_PlatformProcessor_getOrCreateCache = FindMethod(env, c_PlatformProcessor, M_PLATFORM_PROCESSOR_GET_OR_CREATE_CACHE);
+ m_PlatformProcessor_destroyCache = FindMethod(env, c_PlatformProcessor, M_PLATFORM_PROCESSOR_DESTROY_CACHE);
m_PlatformProcessor_affinity = FindMethod(env, c_PlatformProcessor, M_PLATFORM_PROCESSOR_AFFINITY);
m_PlatformProcessor_dataStreamer = FindMethod(env, c_PlatformProcessor, M_PLATFORM_PROCESSOR_DATA_STREAMER);
m_PlatformProcessor_transactions = FindMethod(env, c_PlatformProcessor, M_PLATFORM_PROCESSOR_TRANSACTIONS);
@@ -1144,6 +1146,24 @@ namespace ignite
return ProcessorCache0(obj, name, jvm->GetMembers().m_PlatformProcessor_getOrCreateCache, errInfo);
}
+ void JniContext::ProcessorDestroyCache(jobject obj, const char* name) {
+ ProcessorDestroyCache(obj, name, NULL);
+ }
+
+ void JniContext::ProcessorDestroyCache(jobject obj, const char* name, JniErrorInfo* errInfo)
+ {
+ JNIEnv* env = Attach();
+
+ jstring name0 = name != NULL ? env->NewStringUTF(name) : NULL;
+
+ env->CallVoidMethod(obj, jvm->GetMembers().m_PlatformProcessor_destroyCache, name0);
+
+ if (name0)
+ env->DeleteLocalRef(name0);
+
+ ExceptionCheck(env, errInfo);
+ }
+
jobject JniContext::ProcessorAffinity(jobject obj, const char* name) {
JNIEnv* env = Attach();
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/CacheAbstractTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/CacheAbstractTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/CacheAbstractTest.cs
index ce15739..57e4949 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/CacheAbstractTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/CacheAbstractTest.cs
@@ -3076,6 +3076,25 @@ namespace Apache.Ignite.Core.Tests.Cache
}
[Test]
+ public void TestDestroy()
+ {
+ var cacheName = "template" + Guid.NewGuid();
+
+ var ignite = GetIgnite(0);
+
+ var cache = ignite.CreateCache<int, int>(cacheName);
+
+ Assert.IsNotNull(ignite.GetCache<int, int>(cacheName));
+
+ ignite.DestroyCache(cache.Name);
+
+ var ex = Assert.Throws<ArgumentException>(() => ignite.GetCache<int, int>(cacheName));
+
+ Assert.IsTrue(ex.Message.StartsWith("Cache doesn't exist"));
+ }
+
+
+ [Test]
public void TestIndexer()
{
var cache = Cache();
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/dotnet/Apache.Ignite.Core/IIgnite.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/IIgnite.cs b/modules/platforms/dotnet/Apache.Ignite.Core/IIgnite.cs
index 2f82756..a85e24c 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/IIgnite.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/IIgnite.cs
@@ -102,6 +102,13 @@ namespace Apache.Ignite.Core
ICache<TK, TV> CreateCache<TK, TV>(string name);
/// <summary>
+ /// Destroys dynamically created (with <see cref="CreateCache{TK,TV}"/> or
+ /// <see cref="GetOrCreateCache{TK,TV}"/>) cache.
+ /// </summary>
+ /// <param name="name">The name of the cache to stop.</param>
+ void DestroyCache(string name);
+
+ /// <summary>
/// Gets a new instance of data streamer associated with given cache name. Data streamer
/// is responsible for loading external data into Ignite. For more information
/// refer to <see cref="IDataStreamer{K,V}"/> documentation.
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Ignite.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Ignite.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Ignite.cs
index 2fcada3..ffc8be8 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Ignite.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Ignite.cs
@@ -344,6 +344,12 @@ namespace Apache.Ignite.Core.Impl
return Cache<TK, TV>(UU.ProcessorCreateCache(_proc, name));
}
+ /** <inheritdoc /> */
+ public void DestroyCache(string name)
+ {
+ UU.ProcessorDestroyCache(_proc, name);
+ }
+
/// <summary>
/// Gets cache from specified native cache object.
/// </summary>
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/dotnet/Apache.Ignite.Core/Impl/IgniteProxy.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/IgniteProxy.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/IgniteProxy.cs
index 36aac1a..16062e2 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/IgniteProxy.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/IgniteProxy.cs
@@ -238,6 +238,12 @@ namespace Apache.Ignite.Core.Impl
}
/** <inheritdoc /> */
+ public void DestroyCache(string name)
+ {
+ _ignite.DestroyCache(name);
+ }
+
+ /** <inheritdoc /> */
public IClusterNode GetLocalNode()
{
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
index 5e54a4c..70ad733 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
@@ -55,6 +55,9 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
[DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteProcessorGetOrCreateCache")]
public static extern void* ProcessorGetOrCreateCache(void* ctx, void* obj, sbyte* name);
+ [DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteProcessorDestroyCache")]
+ public static extern void* ProcessorDestroyCache(void* ctx, void* obj, sbyte* name);
+
[DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteProcessorAffinity")]
public static extern void* ProcessorAffinity(void* ctx, void* obj, sbyte* name);
http://git-wip-us.apache.org/repos/asf/ignite/blob/9a996332/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
index e9800ee..7b12010 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
@@ -163,6 +163,22 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
}
}
+ internal static IUnmanagedTarget ProcessorDestroyCache(IUnmanagedTarget target, string name)
+ {
+ sbyte* name0 = IgniteUtils.StringToUtf8Unmanaged(name);
+
+ try
+ {
+ void* res = JNI.ProcessorDestroyCache(target.Context, target.Target, name0);
+
+ return target.ChangeTarget(res);
+ }
+ finally
+ {
+ Marshal.FreeHGlobal(new IntPtr(name0));
+ }
+ }
+
internal static IUnmanagedTarget ProcessorAffinity(IUnmanagedTarget target, string name)
{
sbyte* name0 = IgniteUtils.StringToUtf8Unmanaged(name);
[17/28] ignite git commit: ignite-1811 Optimized cache 'get' on
affinity node.
Posted by vo...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheGetRestartTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheGetRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheGetRestartTest.java
new file mode 100644
index 0000000..b14109b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheGetRestartTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.distributed;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.cache.CacheRebalanceMode.ASYNC;
+
+/**
+ *
+ */
+public class IgniteCacheGetRestartTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final long TEST_TIME = 60_000;
+
+ /** */
+ private static final int SRVS = 3;
+
+ /** */
+ private static final int CLIENTS = 1;
+
+ /** */
+ private static final int KEYS = 100_000;
+
+ /** */
+ private ThreadLocal<Boolean> client = new ThreadLocal<>();
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ Boolean clientMode = client.get();
+
+ if (clientMode != null) {
+ cfg.setClientMode(clientMode);
+
+ client.remove();
+ }
+
+ cfg.setConsistentId(gridName);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGrids(SRVS);
+
+ for (int i = 0; i < CLIENTS; i++) {
+ client.set(true);
+
+ Ignite client = startGrid(SRVS);
+
+ assertTrue(client.configuration().isClientMode());
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return TEST_TIME + 60_000;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetRestartReplicated() throws Exception {
+ CacheConfiguration<Object, Object> cache = cacheConfiguration(REPLICATED, 0, false);
+
+ checkRestart(cache, 3);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetRestartPartitioned1() throws Exception {
+ CacheConfiguration<Object, Object> cache = cacheConfiguration(PARTITIONED, 1, false);
+
+ checkRestart(cache, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetRestartPartitioned2() throws Exception {
+ CacheConfiguration<Object, Object> cache = cacheConfiguration(PARTITIONED, 2, false);
+
+ checkRestart(cache, 2);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetRestartPartitionedNearEnabled() throws Exception {
+ CacheConfiguration<Object, Object> cache = cacheConfiguration(PARTITIONED, 1, true);
+
+ checkRestart(cache, 1);
+ }
+
+ /**
+ * @param ccfg Cache configuration.
+ * @param restartCnt Number of nodes to restart.
+ * @throws Exception If failed.
+ */
+ private void checkRestart(final CacheConfiguration ccfg, final int restartCnt) throws Exception {
+ ignite(0).createCache(ccfg);
+
+ try {
+ if (ccfg.getNearConfiguration() != null)
+ ignite(SRVS).createNearCache(ccfg.getName(), new NearCacheConfiguration<>());
+
+ try (IgniteDataStreamer<Object, Object> streamer = ignite(0).dataStreamer(ccfg.getName())) {
+ for (int i = 0; i < KEYS; i++)
+ streamer.addData(i, i);
+ }
+
+ final long stopTime = U.currentTimeMillis() + TEST_TIME;
+
+ final AtomicInteger nodeIdx = new AtomicInteger();
+
+ IgniteInternalFuture<?> fut1 = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ Ignite ignite = ignite(nodeIdx.getAndIncrement());
+
+ log.info("Check get [node=" + ignite.name() +
+ ", client=" + ignite.configuration().isClientMode() + ']');
+
+ IgniteCache<Object, Object> cache = ignite.cache(ccfg.getName());
+
+ while (U.currentTimeMillis() < stopTime)
+ checkGet(cache);
+
+ return null;
+ }
+ }, SRVS + CLIENTS, "get-thread");
+
+ final AtomicInteger restartNodeIdx = new AtomicInteger(SRVS + CLIENTS);
+
+ final AtomicBoolean clientNode = new AtomicBoolean();
+
+ IgniteInternalFuture<?> fut2 = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ int nodeIdx = restartNodeIdx.getAndIncrement();
+
+ boolean clientMode = clientNode.compareAndSet(false, true);
+
+ while (U.currentTimeMillis() < stopTime) {
+ if (clientMode)
+ client.set(true);
+
+ log.info("Restart node [node=" + nodeIdx + ", client=" + clientMode + ']');
+
+ Ignite ignite = startGrid(nodeIdx);
+
+ IgniteCache<Object, Object> cache;
+
+ if (clientMode && ccfg.getNearConfiguration() != null)
+ cache = ignite.createNearCache(ccfg.getName(), new NearCacheConfiguration<>());
+ else
+ cache = ignite.cache(ccfg.getName());
+
+ checkGet(cache);
+
+ IgniteInternalFuture<?> syncFut = ((IgniteCacheProxy)cache).context().preloader().syncFuture();
+
+ while (!syncFut.isDone())
+ checkGet(cache);
+
+ checkGet(cache);
+
+ stopGrid(nodeIdx);
+ }
+
+ return null;
+ }
+ }, restartCnt + 1, "restart-thread");
+
+ fut1.get();
+ fut2.get();
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param cache Cache.
+ */
+ private void checkGet(IgniteCache<Object, Object> cache) {
+ for (int i = 0; i < KEYS; i++)
+ assertEquals(i, cache.get(i));
+
+ Set<Integer> keys = new HashSet<>();
+
+ for (int i = 0; i < KEYS; i++) {
+ keys.add(i);
+
+ if (keys.size() == 100) {
+ Map<Object, Object> vals = cache.getAll(keys);
+
+ for (Object key : keys)
+ assertEquals(key, vals.get(key));
+
+ keys.clear();
+ }
+ }
+ }
+
+ /**
+ * @param cacheMode Cache mode.
+ * @param backups Number of backups.
+ * @param near If {@code true} near cache is enabled.
+ * @return Cache configuration.
+ */
+ private CacheConfiguration<Object, Object> cacheConfiguration(CacheMode cacheMode, int backups, boolean near) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+ ccfg.setCacheMode(cacheMode);
+
+ if (cacheMode != REPLICATED)
+ ccfg.setBackups(backups);
+
+ if (near)
+ ccfg.setNearConfiguration(new NearCacheConfiguration<>());
+
+ ccfg.setRebalanceMode(ASYNC);
+
+ return ccfg;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
new file mode 100644
index 0000000..af018cc
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
@@ -0,0 +1,427 @@
+/*
+ * 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.distributed;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.cache.Cache;
+import javax.cache.configuration.Factory;
+import javax.cache.integration.CacheLoaderException;
+import javax.cache.integration.CacheWriterException;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.store.CacheStore;
+import org.apache.ignite.cache.store.CacheStoreAdapter;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessageV2;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+
+/**
+ *
+ */
+public class IgniteCacheReadFromBackupTest extends GridCommonAbstractTest {
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int NODES = 4;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
+
+ cfg.setCommunicationSpi(commSpi);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGridsMultiThreaded(NODES);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetFromBackupStoreReadThroughEnabled() throws Exception {
+ for (CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ ccfg.setCacheStoreFactory(new TestStoreFactory());
+ ccfg.setReadThrough(true);
+
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ log.info("Check node: " + ignite.name());
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ TestRecordingCommunicationSpi spi = recordGetRequests(ignite, near);
+
+ Integer key = backupKey(cache);
+
+ assertNull(cache.get(key));
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertEquals(1, msgs.size());
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetFromBackupStoreReadThroughDisabled() throws Exception {
+ for (CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ ccfg.setCacheStoreFactory(new TestStoreFactory());
+ ccfg.setReadThrough(false);
+
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ checkLocalRead(NODES, ccfg);
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetFromPrimaryPreloadInProgress() throws Exception {
+ for (final CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ Map<Ignite, Integer> backupKeys = new HashMap<>();
+ Map<Ignite, Integer> nearKeys = new HashMap<>();
+
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ backupKeys.put(ignite, backupKey(cache));
+
+ if (ccfg.getCacheMode() == PARTITIONED)
+ nearKeys.put(ignite, nearKey(cache));
+
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ spi.blockMessages(new IgnitePredicate<GridIoMessage>() {
+ @Override public boolean apply(GridIoMessage ioMsg) {
+ if (!ioMsg.message().getClass().equals(GridDhtPartitionSupplyMessageV2.class))
+ return false;
+
+ GridDhtPartitionSupplyMessageV2 msg = (GridDhtPartitionSupplyMessageV2)ioMsg.message();
+
+ return msg.cacheId() == CU.cacheId(ccfg.getName());
+ }
+ });
+ }
+
+ try (Ignite newNode = startGrid(NODES)) {
+ IgniteCache<Integer, Integer> cache = newNode.cache(ccfg.getName());
+
+ TestRecordingCommunicationSpi newNodeSpi = recordGetRequests(newNode, near);
+
+ Integer key = backupKey(cache);
+
+ assertNull(cache.get(key));
+
+ List<Object> msgs = newNodeSpi.recordedMessages();
+
+ assertEquals(1, msgs.size());
+
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ log.info("Check node: " + ignite.name());
+
+ checkLocalRead(ignite, ccfg, backupKeys.get(ignite), nearKeys.get(ignite));
+ }
+
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ spi.stopBlock();
+ }
+
+ awaitPartitionMapExchange();
+
+ checkLocalRead(NODES + 1, ccfg);
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNoPrimaryReadPreloadFinished() throws Exception {
+ for (CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ checkLocalRead(NODES, ccfg);
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @param nodes Number of nodes.
+ * @param ccfg Cache configuration.
+ * @throws Exception If failed.
+ */
+ private void checkLocalRead(int nodes, CacheConfiguration<Object, Object> ccfg) throws Exception {
+ for (int i = 0; i < nodes; i++) {
+ Ignite ignite = ignite(i);
+
+ log.info("Check node: " + ignite.name());
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ List<Integer> backupKeys = backupKeys(cache, 2, 0);
+
+ Integer backupKey = backupKeys.get(0);
+
+ Integer nearKey = ccfg.getCacheMode() == PARTITIONED ? nearKey(cache) : null;
+
+ checkLocalRead(ignite, ccfg, backupKey, nearKey);
+
+ Set<Integer> keys = new HashSet<>(backupKeys);
+
+ Map<Integer, Integer> vals = cache.getAll(keys);
+
+ for (Integer key : keys)
+ assertNull(vals.get(key));
+
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertEquals(0, msgs.size());
+ }
+ }
+
+ /**
+ * @param ignite Node.
+ * @param ccfg Cache configuration.
+ * @param backupKey Backup key.
+ * @param nearKey Near key.
+ * @throws Exception If failed.
+ */
+ private void checkLocalRead(Ignite ignite,
+ CacheConfiguration<Object, Object> ccfg,
+ Integer backupKey,
+ Integer nearKey) throws Exception {
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ TestRecordingCommunicationSpi spi = recordGetRequests(ignite, ccfg.getNearConfiguration() != null);
+
+ List<Object> msgs;
+
+ if (nearKey != null) {
+ assertNull(cache.get(nearKey));
+
+ msgs = spi.recordedMessages();
+
+ assertEquals(1, msgs.size());
+ }
+
+ assertNull(cache.get(backupKey));
+
+ msgs = spi.recordedMessages();
+
+ assertTrue(msgs.isEmpty());
+ }
+
+ /**
+ * @param ignite Node.
+ * @param near Near cache flag.
+ * @return Communication SPI.
+ */
+ private TestRecordingCommunicationSpi recordGetRequests(Ignite ignite, boolean near) {
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ spi.record(near ? GridNearGetRequest.class : GridNearSingleGetRequest.class);
+
+ return spi;
+ }
+
+ /**
+ * @return Cache configurations to test.
+ */
+ private List<CacheConfiguration<Object, Object>> cacheConfigurations() {
+ List<CacheConfiguration<Object, Object>> ccfgs = new ArrayList<>();
+
+ ccfgs.add(cacheConfiguration(REPLICATED, ATOMIC, 0, false));
+ ccfgs.add(cacheConfiguration(REPLICATED, TRANSACTIONAL, 0, false));
+
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, 1, false));
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, 1, true));
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, 2, false));
+
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, 1, false));
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, 1, true));
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, 2, false));
+
+ return ccfgs;
+ }
+
+ /**
+ * @param cacheMode Cache mode.
+ * @param atomicityMode Cache atomicity mode.
+ * @param backups Number of backups.
+ * @param nearEnabled {@code True} if near cache should be enabled.
+ * @return Cache configuration.
+ */
+ private CacheConfiguration<Object, Object> cacheConfiguration(CacheMode cacheMode,
+ CacheAtomicityMode atomicityMode,
+ int backups,
+ boolean nearEnabled) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+ ccfg.setCacheMode(cacheMode);
+ ccfg.setAtomicityMode(atomicityMode);
+
+ if (cacheMode != REPLICATED) {
+ ccfg.setBackups(backups);
+
+ if (nearEnabled)
+ ccfg.setNearConfiguration(new NearCacheConfiguration<>());
+ }
+
+ return ccfg;
+ }
+
+ /**
+ *
+ */
+ private static class TestStoreFactory implements Factory<CacheStore<Object, Object>> {
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public CacheStore<Object, Object> create() {
+ return new CacheStoreAdapter() {
+ @Override public Object load(Object key) throws CacheLoaderException {
+ return null;
+ }
+
+ @Override public void write(Cache.Entry entry) throws CacheWriterException {
+ // No-op.
+ }
+
+ @Override public void delete(Object key) throws CacheWriterException {
+ // No-op.
+ }
+ };
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
index 42b5ee3..48fc961 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
@@ -21,26 +21,19 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.spi.IgniteSpiException;
-import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.transactions.Transaction;
-import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
@@ -69,7 +62,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
cfg.setClientMode(client);
- TestCommunicationSpi commSpi = new TestCommunicationSpi();
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
cfg.setCommunicationSpi(commSpi);
@@ -156,7 +149,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
Ignite node = cache.unwrap(Ignite.class);
- TestCommunicationSpi spi = (TestCommunicationSpi)node.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi = (TestRecordingCommunicationSpi)node.configuration().getCommunicationSpi();
spi.record(GridNearSingleGetRequest.class);
@@ -164,17 +157,24 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
assertNotSame(node, primary);
- TestCommunicationSpi primarySpi = (TestCommunicationSpi)primary.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi primarySpi =
+ (TestRecordingCommunicationSpi)primary.configuration().getCommunicationSpi();
primarySpi.record(GridNearSingleGetResponse.class);
assertNull(cache.get(key));
- checkMessages(spi, primarySpi);
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
assertFalse(cache.containsKey(key));
- checkMessages(spi, primarySpi);
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
cache.put(key, 1);
@@ -201,7 +201,10 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
tx.commit();
}
- checkMessages(spi, primarySpi);
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
try (Transaction tx = node.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
assertFalse(cache.containsKey(key));
@@ -209,7 +212,10 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
tx.commit();
}
- checkMessages(spi, primarySpi);
+ if (backup)
+ checkNoMessages(spi, primarySpi);
+ else
+ checkMessages(spi, primarySpi);
cache.put(key, 1);
@@ -241,7 +247,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
* @param spi Near node SPI.
* @param primarySpi Primary node SPI.
*/
- private void checkMessages(TestCommunicationSpi spi, TestCommunicationSpi primarySpi) {
+ private void checkMessages(TestRecordingCommunicationSpi spi, TestRecordingCommunicationSpi primarySpi) {
List<Object> msgs = spi.recordedMessages();
assertEquals(1, msgs.size());
@@ -257,7 +263,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
* @param spi Near node SPI.
* @param primarySpi Primary node SPI.
*/
- private void checkNoMessages(TestCommunicationSpi spi, TestCommunicationSpi primarySpi) {
+ private void checkNoMessages(TestRecordingCommunicationSpi spi, TestRecordingCommunicationSpi primarySpi) {
List<Object> msgs = spi.recordedMessages();
assertEquals(0, msgs.size());
@@ -306,52 +312,4 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
return ccfg;
}
-
- /**
- *
- */
- private static class TestCommunicationSpi extends TcpCommunicationSpi {
- /** */
- private Class<?> recordCls;
-
- /** */
- private List<Object> recordedMsgs = new ArrayList<>();
-
- /** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
- throws IgniteSpiException {
- if (msg instanceof GridIoMessage) {
- Object msg0 = ((GridIoMessage)msg).message();
-
- synchronized (this) {
- if (recordCls != null && msg0.getClass().equals(recordCls))
- recordedMsgs.add(msg0);
- }
- }
-
- super.sendMessage(node, msg, ackC);
- }
-
- /**
- * @param recordCls Message class to record.
- */
- void record(@Nullable Class<?> recordCls) {
- synchronized (this) {
- this.recordCls = recordCls;
- }
- }
-
- /**
- * @return Recorded messages.
- */
- List<Object> recordedMessages() {
- synchronized (this) {
- List<Object> msgs = recordedMsgs;
-
- recordedMsgs = new ArrayList<>();
-
- return msgs;
- }
- }
- }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
index 68cac17..94613db 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
@@ -300,6 +300,7 @@ public class IgniteCrossCacheTxStoreSelfTest extends GridCommonAbstractTest {
throws CacheLoaderException {
}
+ /** {@inheritDoc} */
@Override public void sessionEnd(boolean commit) throws CacheWriterException {
evts.offer("sessionEnd " + commit);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
index fd94150..0666349 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
@@ -17,25 +17,17 @@
package org.apache.ignite.internal.processors.cache.distributed.dht;
-import java.util.Collection;
-import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.CacheRebalanceMode;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.spi.IgniteSpiException;
-import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -79,7 +71,11 @@ public class GridCacheDhtPreloadMessageCountTest extends GridCommonAbstractTest
c.setDiscoverySpi(disco);
c.setCacheConfiguration(cc);
- c.setCommunicationSpi(new TestCommunicationSpi());
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
+
+ commSpi.record(GridDhtPartitionsSingleMessage.class);
+
+ c.setCommunicationSpi(commSpi);
return c;
}
@@ -110,11 +106,13 @@ public class GridCacheDhtPreloadMessageCountTest extends GridCommonAbstractTest
IgniteCache<String, Integer> c1 = g1.cache(null);
IgniteCache<String, Integer> c2 = g2.cache(null);
- TestCommunicationSpi spi0 = (TestCommunicationSpi)g0.configuration().getCommunicationSpi();
- TestCommunicationSpi spi1 = (TestCommunicationSpi)g1.configuration().getCommunicationSpi();
- TestCommunicationSpi spi2 = (TestCommunicationSpi)g2.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi0 = (TestRecordingCommunicationSpi)g0.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi1 = (TestRecordingCommunicationSpi)g1.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi2 = (TestRecordingCommunicationSpi)g2.configuration().getCommunicationSpi();
- info(spi0.sentMessages().size() + " " + spi1.sentMessages().size() + " " + spi2.sentMessages().size());
+ info(spi0.recordedMessages().size() + " " +
+ spi1.recordedMessages().size() + " " +
+ spi2.recordedMessages().size());
checkCache(c0, cnt);
checkCache(c1, cnt);
@@ -137,40 +135,4 @@ public class GridCacheDhtPreloadMessageCountTest extends GridCommonAbstractTest
assertEquals(Integer.valueOf(i), c.localPeek(key, CachePeekMode.ONHEAP));
}
}
-
- /**
- * Communication SPI that will count single partition update messages.
- */
- private static class TestCommunicationSpi extends TcpCommunicationSpi {
- /** Recorded messages. */
- private Collection<GridDhtPartitionsSingleMessage> sentMsgs = new ConcurrentLinkedQueue<>();
-
- /** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackClosure)
- throws IgniteSpiException {
- recordMessage((GridIoMessage)msg);
-
- super.sendMessage(node, msg, ackClosure);
- }
-
- /**
- * @return Collection of sent messages.
- */
- public Collection<GridDhtPartitionsSingleMessage> sentMessages() {
- return sentMsgs;
- }
-
- /**
- * Adds message to a list if message is of correct type.
- *
- * @param msg Message.
- */
- private void recordMessage(GridIoMessage msg) {
- if (msg.message() instanceof GridDhtPartitionsSingleMessage) {
- GridDhtPartitionsSingleMessage partSingleMsg = (GridDhtPartitionsSingleMessage)msg.message();
-
- sentMsgs.add(partSingleMsg);
- }
- }
- }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
index 7bd845a..3e6a245 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
@@ -107,7 +107,11 @@ public class GridCacheGetStoreErrorSelfTest extends GridCommonAbstractTest {
checkGetError(false, LOCAL);
}
- /** @throws Exception If failed. */
+ /**
+ * @param nearEnabled Near cache flag.
+ * @param cacheMode Cache mode.
+ * @throws Exception If failed.
+ */
private void checkGetError(boolean nearEnabled, CacheMode cacheMode) throws Exception {
this.nearEnabled = nearEnabled;
this.cacheMode = cacheMode;
@@ -147,14 +151,17 @@ public class GridCacheGetStoreErrorSelfTest extends GridCommonAbstractTest {
*/
@SuppressWarnings("PublicInnerClass")
public static class TestStore extends CacheStoreAdapter<Object, Object> {
+ /** {@inheritDoc} */
@Override public Object load(Object key) {
throw new IgniteException("Failed to get key from store: " + key);
}
+ /** {@inheritDoc} */
@Override public void write(Cache.Entry<?, ?> entry) {
// No-op.
}
+ /** {@inheritDoc} */
@Override public void delete(Object key) {
// No-op.
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedNodeRestartTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedNodeRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedNodeRestartTest.java
index 684d6e4..f32a5f7 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedNodeRestartTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedNodeRestartTest.java
@@ -23,7 +23,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheAbstractNodeRestartSelfTest;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
-import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_ASYNC;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
/**
@@ -46,7 +46,7 @@ public class GridCachePartitionedNodeRestartTest extends GridCacheAbstractNodeRe
cc.setName(CACHE_NAME);
cc.setAtomicityMode(atomicityMode());
cc.setCacheMode(PARTITIONED);
- cc.setWriteSynchronizationMode(FULL_ASYNC);
+ cc.setWriteSynchronizationMode(FULL_SYNC);
cc.setNearConfiguration(null);
cc.setStartSize(20);
cc.setRebalanceMode(rebalancMode);
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedOptimisticTxNodeRestartTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedOptimisticTxNodeRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedOptimisticTxNodeRestartTest.java
index a458aa7..ab7caad 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedOptimisticTxNodeRestartTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedOptimisticTxNodeRestartTest.java
@@ -25,7 +25,7 @@ import org.apache.ignite.internal.processors.cache.distributed.GridCacheAbstract
import org.apache.ignite.transactions.TransactionConcurrency;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
-import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_ASYNC;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
/**
@@ -53,7 +53,7 @@ public class GridCachePartitionedOptimisticTxNodeRestartTest extends GridCacheAb
cc.setName(CACHE_NAME);
cc.setCacheMode(PARTITIONED);
- cc.setWriteSynchronizationMode(FULL_ASYNC);
+ cc.setWriteSynchronizationMode(FULL_SYNC);
cc.setStartSize(20);
cc.setRebalanceMode(rebalancMode);
cc.setRebalanceBatchSize(rebalancBatchSize);
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
index de87e99..0513786 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheRestartTestSuite2.java
@@ -23,6 +23,7 @@ import org.apache.ignite.internal.processors.cache.IgniteCacheAtomicPutAllFailov
import org.apache.ignite.internal.processors.cache.IgniteCachePutAllRestartTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteBinaryMetadataUpdateNodeRestartTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheAtomicNodeRestartTest;
+import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheGetRestartTest;
import org.apache.ignite.internal.processors.cache.distributed.replicated.IgniteCacheAtomicReplicatedNodeRestartSelfTest;
/**
@@ -45,6 +46,8 @@ public class IgniteCacheRestartTestSuite2 extends TestSuite {
suite.addTestSuite(IgniteBinaryMetadataUpdateNodeRestartTest.class);
+ suite.addTestSuite(IgniteCacheGetRestartTest.class);
+
return suite;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index 04d0881..68e52df 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -84,6 +84,7 @@ import org.apache.ignite.internal.processors.cache.distributed.CacheGetFutureHan
import org.apache.ignite.internal.processors.cache.distributed.CacheNoValueClassOnServerNodeTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheCreatePutMultiNodeSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheCreatePutTest;
+import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheReadFromBackupTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheSingleGetMessageTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridCacheDhtTxPreloadSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteCacheLockFailoverSelfTest;
@@ -289,6 +290,7 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(CacheGetFutureHangsSelfTest.class);
suite.addTestSuite(IgniteCacheSingleGetMessageTest.class);
+ suite.addTestSuite(IgniteCacheReadFromBackupTest.class);
suite.addTestSuite(IgniteCacheGetCustomCollectionsSelfTest.class);
suite.addTestSuite(IgniteCacheLoadRebalanceEvictionSelfTest.class);
[09/28] ignite git commit: IGNITE-2043 .NET: Fix ServiceExample
xmldoc. This closes #296.
Posted by vo...@apache.org.
IGNITE-2043 .NET: Fix ServiceExample xmldoc. This closes #296.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2e1d50d5
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2e1d50d5
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2e1d50d5
Branch: refs/heads/ignite-2236
Commit: 2e1d50d59a716a23a9f36e99af78d3fa4308e910
Parents: add8379
Author: Pavel Tupitsyn <pt...@gridgain.com>
Authored: Mon Jan 18 12:42:46 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 12:42:46 2016 +0300
----------------------------------------------------------------------
.../examples/Apache.Ignite.Examples/Services/ServicesExample.cs | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2e1d50d5/modules/platforms/dotnet/examples/Apache.Ignite.Examples/Services/ServicesExample.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/examples/Apache.Ignite.Examples/Services/ServicesExample.cs b/modules/platforms/dotnet/examples/Apache.Ignite.Examples/Services/ServicesExample.cs
index 7ff44b4..d513305 100644
--- a/modules/platforms/dotnet/examples/Apache.Ignite.Examples/Services/ServicesExample.cs
+++ b/modules/platforms/dotnet/examples/Apache.Ignite.Examples/Services/ServicesExample.cs
@@ -33,7 +33,7 @@ namespace Apache.Ignite.Examples.Services
/// <para />
/// This example can be run with standalone Apache Ignite.NET node:
/// 1) Run %IGNITE_HOME%/platforms/dotnet/bin/Apache.Ignite.exe:
- /// Apache.Ignite.exe -IgniteHome="%IGNITE_HOME%" -springConfigUrl=platforms\dotnet\examples\config\example-cache.xml -assembly=[path_to_Apache.Ignite.ExamplesDll.dll]
+ /// Apache.Ignite.exe -IgniteHome="%IGNITE_HOME%" -springConfigUrl=platforms\dotnet\examples\config\example-compute.xml -assembly=[path_to_Apache.Ignite.ExamplesDll.dll]
/// 2) Start example.
/// </summary>
public class ServicesExample
[18/28] ignite git commit: ignite-1811 Optimized cache 'get' on
affinity node.
Posted by vo...@apache.org.
ignite-1811 Optimized cache 'get' on affinity node.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/83b2bf5e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/83b2bf5e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/83b2bf5e
Branch: refs/heads/ignite-2236
Commit: 83b2bf5e1f287dc83343945b0e47b83ee7724a8e
Parents: d85616b
Author: sboikov <sb...@gridgain.com>
Authored: Mon Jan 18 18:05:37 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Mon Jan 18 18:05:37 2016 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheAdapter.java | 30 +-
.../processors/cache/GridCacheContext.java | 33 ++
.../dht/CacheDistributedGetFutureAdapter.java | 28 +-
.../dht/GridClientPartitionTopology.java | 2 +
.../dht/GridDhtPartitionTopologyImpl.java | 27 +-
.../dht/GridPartitionedGetFuture.java | 241 ++++++-----
.../dht/GridPartitionedSingleGetFuture.java | 229 ++++++----
.../dht/atomic/GridDhtAtomicCache.java | 26 ++
.../distributed/near/GridNearGetFuture.java | 267 +++++++-----
.../cache/transactions/IgniteTxManager.java | 18 +-
.../internal/TestRecordingCommunicationSpi.java | 157 +++++++
...idCacheConfigurationConsistencySelfTest.java | 58 +--
.../cache/IgniteCacheNearLockValueSelfTest.java | 62 +--
.../cache/IgniteCacheStoreCollectionTest.java | 12 +
...eDynamicCacheStartNoExchangeTimeoutTest.java | 7 +
...ridCachePartitionNotLoadedEventSelfTest.java | 7 +-
.../IgniteCacheAtomicNodeRestartTest.java | 2 +
...niteCacheClientNodeChangingTopologyTest.java | 4 +-
.../distributed/IgniteCacheGetRestartTest.java | 280 ++++++++++++
.../IgniteCacheReadFromBackupTest.java | 427 +++++++++++++++++++
.../IgniteCacheSingleGetMessageTest.java | 88 +---
.../IgniteCrossCacheTxStoreSelfTest.java | 1 +
.../GridCacheDhtPreloadMessageCountTest.java | 62 +--
.../near/GridCacheGetStoreErrorSelfTest.java | 9 +-
.../GridCachePartitionedNodeRestartTest.java | 4 +-
...ePartitionedOptimisticTxNodeRestartTest.java | 4 +-
.../IgniteCacheRestartTestSuite2.java | 3 +
.../testsuites/IgniteCacheTestSuite4.java | 2 +
28 files changed, 1524 insertions(+), 566 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 5d4c386..2582e6c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -4540,9 +4540,33 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @return Cached value.
* @throws IgniteCheckedException If failed.
*/
- @Nullable public V get(K key, boolean deserializeBinary)
- throws IgniteCheckedException {
- return getAsync(key, deserializeBinary).get();
+ @Nullable public V get(K key, boolean deserializeBinary) throws IgniteCheckedException {
+ checkJta();
+
+ String taskName = ctx.kernalContext().job().currentTaskName();
+
+ return get(key, taskName, deserializeBinary);
+ }
+
+ /**
+ * @param key Key.
+ * @param taskName Task name.
+ * @param deserializeBinary Deserialize binary flag.
+ * @return Cached value.
+ * @throws IgniteCheckedException If failed.
+ */
+ protected V get(
+ final K key,
+ String taskName,
+ boolean deserializeBinary) throws IgniteCheckedException {
+ return getAsync(key,
+ !ctx.config().isReadFromBackup(),
+ /*skip tx*/false,
+ null,
+ taskName,
+ deserializeBinary,
+ false,
+ /*can remap*/true).get();
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index c10ebf3..fc48b9d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -111,6 +111,7 @@ import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_VALUES;
import static org.apache.ignite.cache.CacheRebalanceMode.NONE;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.OWNING;
/**
* Cache context.
@@ -1434,6 +1435,13 @@ public class GridCacheContext<K, V> implements Externalizable {
}
/**
+ * @return {@code True} if store and read-through mode are enabled in configuration.
+ */
+ public boolean readThroughConfigured() {
+ return store().configured() && cacheCfg.isReadThrough();
+ }
+
+ /**
* @return {@code True} if {@link CacheConfiguration#isLoadPreviousValue()} flag is set.
*/
public boolean loadPreviousValue() {
@@ -1961,6 +1969,31 @@ public class GridCacheContext<K, V> implements Externalizable {
});
}
+ /**
+ * @param part Partition.
+ * @param affNodes Affinity nodes.
+ * @param topVer Topology version.
+ * @return {@code True} if cache 'get' operation is allowed to get entry locally.
+ */
+ public boolean allowFastLocalRead(int part, List<ClusterNode> affNodes, AffinityTopologyVersion topVer) {
+ return affinityNode() && rebalanceEnabled() && hasPartition(part, affNodes, topVer);
+ }
+
+ /**
+ * @param part Partition.
+ * @param affNodes Affinity nodes.
+ * @param topVer Topology version.
+ * @return {@code True} if partition is available locally.
+ */
+ private boolean hasPartition(int part, List<ClusterNode> affNodes, AffinityTopologyVersion topVer) {
+ assert affinityNode();
+
+ GridDhtPartitionTopology top = topology();
+
+ return (top.rebalanceFinished(topVer) && (isReplicated() || affNodes.contains(locNode)))
+ || (top.partitionState(localNodeId(), part) == OWNING);
+ }
+
/** {@inheritDoc} */
@Override public void writeExternal(ObjectOutput out) throws IOException {
U.writeString(out, gridName());
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
index c43cce9..40eec63 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
@@ -24,6 +24,7 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
@@ -39,6 +40,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_NEAR_GET_MAX_REMAPS;
import static org.apache.ignite.IgniteSystemProperties.getInteger;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.OWNING;
/**
*
@@ -168,14 +170,11 @@ public abstract class CacheDistributedGetFutureAdapter<K, V> extends GridCompoun
/**
* Affinity node to send get request to.
*
- * @param key Key to get.
- * @param topVer Topology version.
+ * @param affNodes All affinity nodes.
* @return Affinity node to get key from.
*/
- protected final ClusterNode affinityNode(KeyCacheObject key, AffinityTopologyVersion topVer) {
+ protected final ClusterNode affinityNode(List<ClusterNode> affNodes) {
if (!canRemap) {
- List<ClusterNode> affNodes = cctx.affinity().nodes(key, topVer);
-
for (ClusterNode node : affNodes) {
if (cctx.discovery().alive(node))
return node;
@@ -184,6 +183,23 @@ public abstract class CacheDistributedGetFutureAdapter<K, V> extends GridCompoun
return null;
}
else
- return cctx.affinity().primary(key, topVer);
+ return affNodes.get(0);
+ }
+
+ /**
+ * @param part Partition.
+ * @return {@code True} if partition is in owned state.
+ */
+ protected final boolean partitionOwned(int part) {
+ return cctx.topology().partitionState(cctx.localNodeId(), part) == OWNING;
+ }
+
+ /**
+ * @param topVer Topology version.
+ * @return Exception.
+ */
+ protected final ClusterTopologyServerNotFoundException serverNotFoundError(AffinityTopologyVersion topVer) {
+ return new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
+ "(all partition nodes left the grid) [topVer=" + topVer + ", cache=" + cctx.name() + ']');
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index 8aef5ad..dcfc038 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -882,6 +882,8 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public boolean rebalanceFinished(AffinityTopologyVersion topVer) {
+ assert false : "Should not be called on non-affinity node";
+
return false;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index a0709c5..2ab8a12 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@ -88,7 +88,7 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
private GridDhtPartitionExchangeId lastExchangeId;
/** */
- private AffinityTopologyVersion topVer = AffinityTopologyVersion.NONE;
+ private volatile AffinityTopologyVersion topVer = AffinityTopologyVersion.NONE;
/** */
private volatile boolean stopping;
@@ -136,9 +136,9 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
topReadyFut = null;
- topVer = AffinityTopologyVersion.NONE;
-
rebalancedTopVer = AffinityTopologyVersion.NONE;
+
+ topVer = AffinityTopologyVersion.NONE;
}
finally {
lock.writeLock().unlock();
@@ -223,13 +223,13 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
this.stopping = stopping;
- topVer = exchId.topologyVersion();
-
updateSeq.setIfGreater(updSeq);
topReadyFut = exchFut;
rebalancedTopVer = AffinityTopologyVersion.NONE;
+
+ topVer = exchId.topologyVersion();
}
finally {
lock.writeLock().unlock();
@@ -238,17 +238,12 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public AffinityTopologyVersion topologyVersion() {
- lock.readLock().lock();
+ AffinityTopologyVersion topVer = this.topVer;
- try {
- assert topVer.topologyVersion() > 0 : "Invalid topology version [topVer=" + topVer +
- ", cacheName=" + cctx.name() + ']';
+ assert topVer.topologyVersion() > 0 : "Invalid topology version [topVer=" + topVer +
+ ", cacheName=" + cctx.name() + ']';
- return topVer;
- }
- finally {
- lock.readLock().unlock();
- }
+ return topVer;
}
/** {@inheritDoc} */
@@ -1336,7 +1331,9 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public boolean rebalanceFinished(AffinityTopologyVersion topVer) {
- return topVer.equals(rebalancedTopVer);
+ AffinityTopologyVersion curTopVer = this.topVer;
+
+ return curTopVer.equals(topVer) && curTopVer.equals(rebalancedTopVer);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 19df1c2..1f2d7c5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
@@ -234,15 +235,16 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
Map<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>> mapped,
AffinityTopologyVersion topVer
) {
- if (CU.affinityNodes(cctx, topVer).isEmpty()) {
+ Collection<ClusterNode> cacheNodes = CU.affinityNodes(cctx, topVer);
+
+ if (cacheNodes.isEmpty()) {
onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
"(all partition nodes left the grid) [topVer=" + topVer + ", cache=" + cctx.name() + ']'));
return;
}
- Map<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>> mappings =
- U.newHashMap(CU.affinityNodes(cctx, topVer).size());
+ Map<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>> mappings = U.newHashMap(cacheNodes.size());
final int keysSize = keys.size();
@@ -374,135 +376,160 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
AffinityTopologyVersion topVer,
Map<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>> mapped
) {
- GridDhtCacheAdapter<K, V> colocated = cache();
+ int part = cctx.affinity().partition(key);
- boolean remote = false;
+ List<ClusterNode> affNodes = cctx.affinity().nodes(part, topVer);
- // Allow to get cached value from the local node.
- boolean allowLocRead = (cctx.affinityNode() && !forcePrimary) ||
- cctx.affinity().primary(cctx.localNode(), key, topVer);
+ if (affNodes.isEmpty()) {
+ onDone(serverNotFoundError(topVer));
- while (true) {
- GridCacheEntryEx entry;
+ return false;
+ }
- try {
- if (allowLocRead) {
- try {
- entry = colocated.context().isSwapOrOffheapEnabled() ? colocated.entryEx(key) :
- colocated.peekEx(key);
-
- // If our DHT cache do has value, then we peek it.
- if (entry != null) {
- boolean isNew = entry.isNewLocked();
-
- CacheObject v = null;
- GridCacheVersion ver = null;
-
- if (needVer) {
- T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
- null,
- /*swap*/true,
- /*unmarshal*/true,
- /**update-metrics*/false,
- /*event*/!skipVals,
- subjId,
- null,
- taskName,
- expiryPlc,
- !deserializeBinary);
-
- if (res != null) {
- v = res.get1();
- ver = res.get2();
- }
- }
- else {
- v = entry.innerGet(null,
- /*swap*/true,
- /*read-through*/false,
- /*fail-fast*/true,
- /*unmarshal*/true,
- /**update-metrics*/false,
- /*event*/!skipVals,
- /*temporary*/false,
- subjId,
- null,
- taskName,
- expiryPlc,
- !deserializeBinary);
- }
+ boolean fastLocGet = (!forcePrimary || affNodes.get(0).isLocal()) &&
+ cctx.allowFastLocalRead(part, affNodes, topVer);
- colocated.context().evicts().touch(entry, topVer);
+ if (fastLocGet && localGet(key, part, locVals))
+ return false;
- // Entry was not in memory or in swap, so we remove it from cache.
- if (v == null) {
- if (isNew && entry.markObsoleteIfEmpty(ver))
- colocated.removeIfObsolete(key);
- }
- else {
- if (needVer)
- versionedResult(locVals, key, v, ver);
- else
- cctx.addResult(locVals,
- key,
- v,
- skipVals,
- keepCacheObjects,
- deserializeBinary,
- true);
-
- return false;
- }
- }
- }
- catch (GridDhtInvalidPartitionException ignored) {
- // No-op.
- }
- }
+ ClusterNode node = affinityNode(affNodes);
- ClusterNode node = affinityNode(key, topVer);
+ if (node == null) {
+ onDone(serverNotFoundError(topVer));
- if (node == null) {
- onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
- "(all partition nodes left the grid)."));
+ return false;
+ }
- return false;
- }
+ boolean remote = !node.isLocal();
+
+ LinkedHashMap<KeyCacheObject, Boolean> keys = mapped.get(node);
+
+ if (keys != null && keys.containsKey(key)) {
+ if (REMAP_CNT_UPD.incrementAndGet(this) > MAX_REMAP_CNT) {
+ onDone(new ClusterTopologyCheckedException("Failed to remap key to a new node after " +
+ MAX_REMAP_CNT + " attempts (key got remapped to the same node) [key=" + key + ", node=" +
+ U.toShortString(node) + ", mappings=" + mapped + ']'));
+
+ return false;
+ }
+ }
+
+ LinkedHashMap<KeyCacheObject, Boolean> old = mappings.get(node);
+
+ if (old == null)
+ mappings.put(node, old = new LinkedHashMap<>(3, 1f));
+
+ old.put(key, false);
+
+ return remote;
+ }
- remote = !node.isLocal();
+ /**
+ * @param key Key.
+ * @param part Partition.
+ * @param locVals Local values.
+ * @return {@code True} if there is no need to further search value.
+ */
+ private boolean localGet(KeyCacheObject key, int part, Map<K, V> locVals) {
+ assert cctx.affinityNode() : this;
- LinkedHashMap<KeyCacheObject, Boolean> keys = mapped.get(node);
+ GridDhtCacheAdapter<K, V> cache = cache();
- if (keys != null && keys.containsKey(key)) {
- if (REMAP_CNT_UPD.incrementAndGet(this) > MAX_REMAP_CNT) {
- onDone(new ClusterTopologyCheckedException("Failed to remap key to a new node after " +
- MAX_REMAP_CNT + " attempts (key got remapped to the same node) [key=" + key + ", node=" +
- U.toShortString(node) + ", mappings=" + mapped + ']'));
+ while (true) {
+ GridCacheEntryEx entry;
- return false;
+ try {
+ entry = cache.context().isSwapOrOffheapEnabled() ? cache.entryEx(key) : cache.peekEx(key);
+
+ // If our DHT cache do has value, then we peek it.
+ if (entry != null) {
+ boolean isNew = entry.isNewLocked();
+
+ CacheObject v = null;
+ GridCacheVersion ver = null;
+
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
+ null,
+ /*swap*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ subjId,
+ null,
+ taskName,
+ expiryPlc,
+ !deserializeBinary);
+
+ if (res != null) {
+ v = res.get1();
+ ver = res.get2();
+ }
+ }
+ else {
+ v = entry.innerGet(null,
+ /*swap*/true,
+ /*read-through*/false,
+ /*fail-fast*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ /*temporary*/false,
+ subjId,
+ null,
+ taskName,
+ expiryPlc,
+ !deserializeBinary);
+ }
+
+ cache.context().evicts().touch(entry, topVer);
+
+ // Entry was not in memory or in swap, so we remove it from cache.
+ if (v == null) {
+ if (isNew && entry.markObsoleteIfEmpty(ver))
+ cache.removeIfObsolete(key);
+ }
+ else {
+ if (needVer)
+ versionedResult(locVals, key, v, ver);
+ else {
+ cctx.addResult(locVals,
+ key,
+ v,
+ skipVals,
+ keepCacheObjects,
+ deserializeBinary,
+ true);
+ }
+
+ return true;
}
}
- LinkedHashMap<KeyCacheObject, Boolean> old = mappings.get(node);
+ boolean topStable = cctx.isReplicated() || topVer.equals(cctx.topology().topologyVersion());
- if (old == null)
- mappings.put(node, old = new LinkedHashMap<>(3, 1f));
+ // Entry not found, do not continue search if topology did not change and there is no store.
+ if (!cctx.readThroughConfigured() && (topStable || partitionOwned(part))) {
+ if (!skipVals && cctx.config().isStatisticsEnabled())
+ cache.metrics0().onRead(false);
- old.put(key, false);
+ return true;
+ }
- break;
+ return false;
+ }
+ catch (GridCacheEntryRemovedException ignored) {
+ // No-op, will retry.
+ }
+ catch (GridDhtInvalidPartitionException ignored) {
+ return false;
}
catch (IgniteCheckedException e) {
onDone(e);
- break;
- }
- catch (GridCacheEntryRemovedException ignored) {
- // No-op, will retry.
+ return true;
}
}
-
- return remote;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 29971fd..0c811ae 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -58,6 +58,8 @@ import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.OWNING;
+
/**
*
*/
@@ -319,105 +321,140 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im
* @return Primary node or {@code null} if future was completed.
*/
@Nullable private ClusterNode mapKeyToNode(AffinityTopologyVersion topVer) {
- ClusterNode primary = affinityNode(key, topVer);
+ int part = cctx.affinity().partition(key);
+
+ List<ClusterNode> affNodes = cctx.affinity().nodes(part, topVer);
- if (primary == null) {
- onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
- "(all partition nodes left the grid) [topVer=" + topVer + ", cache=" + cctx.name() + ']'));
+ if (affNodes.isEmpty()) {
+ onDone(serverNotFoundError(topVer));
return null;
}
- boolean allowLocRead = (cctx.affinityNode() && !forcePrimary) || primary.isLocal();
-
- if (allowLocRead) {
- GridDhtCacheAdapter colocated = cctx.dht();
-
- while (true) {
- GridCacheEntryEx entry;
-
- try {
- entry = colocated.context().isSwapOrOffheapEnabled() ? colocated.entryEx(key) :
- colocated.peekEx(key);
-
- // If our DHT cache do has value, then we peek it.
- if (entry != null) {
- boolean isNew = entry.isNewLocked();
-
- CacheObject v = null;
- GridCacheVersion ver = null;
-
- if (needVer) {
- T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
- null,
- /*swap*/true,
- /*unmarshal*/true,
- /**update-metrics*/false,
- /*event*/!skipVals,
- subjId,
- null,
- taskName,
- expiryPlc,
- true);
-
- if (res != null) {
- v = res.get1();
- ver = res.get2();
- }
- }
- else {
- v = entry.innerGet(null,
- /*swap*/true,
- /*read-through*/false,
- /*fail-fast*/true,
- /*unmarshal*/true,
- /**update-metrics*/false,
- /*event*/!skipVals,
- /*temporary*/false,
- subjId,
- null,
- taskName,
- expiryPlc,
- true);
- }
+ boolean fastLocGet = (!forcePrimary || affNodes.get(0).isLocal()) &&
+ cctx.allowFastLocalRead(part, affNodes, topVer);
- colocated.context().evicts().touch(entry, topVer);
+ if (fastLocGet && localGet(topVer, part))
+ return null;
- // Entry was not in memory or in swap, so we remove it from cache.
- if (v == null) {
- if (isNew && entry.markObsoleteIfEmpty(ver))
- colocated.removeIfObsolete(key);
- }
- else {
- if (!skipVals && cctx.config().isStatisticsEnabled())
- cctx.cache().metrics0().onRead(true);
+ ClusterNode affNode = affinityNode(affNodes);
+
+ if (affNode == null) {
+ onDone(serverNotFoundError(topVer));
+
+ return null;
+ }
+
+ return affNode;
+ }
+
+ /**
+ * @param topVer Topology version.
+ * @param part Partition.
+ * @return {@code True} if future completed.
+ */
+ private boolean localGet(AffinityTopologyVersion topVer, int part) {
+ assert cctx.affinityNode() : this;
+
+ GridDhtCacheAdapter colocated = cctx.dht();
- if (!skipVals)
- setResult(v, ver);
- else
- setSkipValueResult(true, ver);
+ while (true) {
+ GridCacheEntryEx entry;
- return null;
+ try {
+ entry = colocated.context().isSwapOrOffheapEnabled() ? colocated.entryEx(key) :
+ colocated.peekEx(key);
+
+ // If our DHT cache do has value, then we peek it.
+ if (entry != null) {
+ boolean isNew = entry.isNewLocked();
+
+ CacheObject v = null;
+ GridCacheVersion ver = null;
+
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(
+ null,
+ /*swap*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ subjId,
+ null,
+ taskName,
+ expiryPlc,
+ true);
+
+ if (res != null) {
+ v = res.get1();
+ ver = res.get2();
}
}
+ else {
+ v = entry.innerGet(null,
+ /*swap*/true,
+ /*read-through*/false,
+ /*fail-fast*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!skipVals,
+ /*temporary*/false,
+ subjId,
+ null,
+ taskName,
+ expiryPlc,
+ true);
+ }
- break;
- }
- catch (GridDhtInvalidPartitionException ignored) {
- break;
- }
- catch (IgniteCheckedException e) {
- onDone(e);
+ colocated.context().evicts().touch(entry, topVer);
+
+ // Entry was not in memory or in swap, so we remove it from cache.
+ if (v == null) {
+ if (isNew && entry.markObsoleteIfEmpty(ver))
+ colocated.removeIfObsolete(key);
+ }
+ else {
+ if (!skipVals && cctx.config().isStatisticsEnabled())
+ cctx.cache().metrics0().onRead(true);
+
+ if (!skipVals)
+ setResult(v, ver);
+ else
+ setSkipValueResult(true, ver);
- return null;
+ return true;
+ }
}
- catch (GridCacheEntryRemovedException ignored) {
- // No-op, will retry.
+
+ boolean topStable = cctx.isReplicated() || topVer.equals(cctx.topology().topologyVersion());
+
+ // Entry not found, complete future with null result if topology did not change and there is no store.
+ if (!cctx.readThroughConfigured() && (topStable || partitionOwned(part))) {
+ if (!skipVals && cctx.config().isStatisticsEnabled())
+ colocated.metrics0().onRead(false);
+
+ if (skipVals)
+ setSkipValueResult(false, null);
+ else
+ setResult(null, null);
+
+ return true;
}
+
+ return false;
}
- }
+ catch (GridCacheEntryRemovedException ignored) {
+ // No-op, will retry.
+ }
+ catch (GridDhtInvalidPartitionException ignored) {
+ return false;
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
- return primary;
+ return true;
+ }
+ }
}
/**
@@ -595,7 +632,7 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im
}
else {
if (!keepCacheObjects) {
- Object res = cctx.unwrapBinaryIfNeeded(val, !deserializeBinary && !skipVals);
+ Object res = cctx.unwrapBinaryIfNeeded(val, !deserializeBinary);
onDone(res);
}
@@ -612,16 +649,30 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im
}
/**
+ * @param part Partition.
+ * @return {@code True} if partition is in owned state.
+ */
+ private boolean partitionOwned(int part) {
+ return cctx.topology().partitionState(cctx.localNodeId(), part) == OWNING;
+ }
+
+ /**
+ * @param topVer Topology version.
+ * @return Exception.
+ */
+ private ClusterTopologyServerNotFoundException serverNotFoundError(AffinityTopologyVersion topVer) {
+ return new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
+ "(all partition nodes left the grid) [topVer=" + topVer + ", cache=" + cctx.name() + ']');
+ }
+
+ /**
* Affinity node to send get request to.
*
- * @param key Key to get.
- * @param topVer Topology version.
+ * @param affNodes All affinity nodes.
* @return Affinity node to get key from.
*/
- private ClusterNode affinityNode(KeyCacheObject key, AffinityTopologyVersion topVer) {
+ @Nullable private ClusterNode affinityNode(List<ClusterNode> affNodes) {
if (!canRemap) {
- List<ClusterNode> affNodes = cctx.affinity().nodes(key, topVer);
-
for (ClusterNode node : affNodes) {
if (cctx.discovery().alive(node))
return node;
@@ -630,7 +681,7 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im
return null;
}
else
- return cctx.affinity().primary(key, topVer);
+ return affNodes.get(0);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 393413e..81fd5d6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -317,6 +317,32 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
/** {@inheritDoc} */
+ @Override protected V get(K key, String taskName, boolean deserializeBinary) throws IgniteCheckedException {
+ ctx.checkSecurity(SecurityPermission.CACHE_READ);
+
+ if (keyCheck)
+ validateCacheKey(key);
+
+ CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+ UUID subjId = ctx.subjectIdPerCall(null, opCtx);
+
+ final ExpiryPolicy expiryPlc = opCtx != null ? opCtx.expiry() : null;
+
+ final boolean skipStore = opCtx != null && opCtx.skipStore();
+
+ return getAsync0(ctx.toCacheKeyObject(key),
+ !ctx.config().isReadFromBackup(),
+ subjId,
+ taskName,
+ deserializeBinary,
+ expiryPlc,
+ false,
+ skipStore,
+ true).get();
+ }
+
+ /** {@inheritDoc} */
@Override protected IgniteInternalFuture<V> getAsync(final K key,
final boolean forcePrimary,
final boolean skipTx,
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index c547a88..9291001 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
@@ -405,10 +406,20 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
Map<ClusterNode, LinkedHashMap<KeyCacheObject, Boolean>> mapped,
Map<KeyCacheObject, GridNearCacheEntry> saved
) {
+ int part = cctx.affinity().partition(key);
+
+ List<ClusterNode> affNodes = cctx.affinity().nodes(part, topVer);
+
+ if (affNodes.isEmpty()) {
+ onDone(serverNotFoundError(topVer));
+
+ return null;
+ }
+
final GridNearCacheAdapter near = cache();
// Allow to get cached value from the local node.
- boolean allowLocRead = !forcePrimary || cctx.affinity().primary(cctx.localNode(), key, topVer);
+ boolean allowLocRead = !forcePrimary || cctx.localNode().equals(affNodes.get(0));
while (true) {
GridNearCacheEntry entry = allowLocRead ? (GridNearCacheEntry)near.peekEx(key) : null;
@@ -456,124 +467,23 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
}
- ClusterNode affNode = null;
-
- if (v == null && allowLocRead && cctx.affinityNode()) {
- GridDhtCacheAdapter<K, V> dht = cache().dht();
-
- GridCacheEntryEx dhtEntry = null;
-
- try {
- dhtEntry = dht.context().isSwapOrOffheapEnabled() ? dht.entryEx(key) : dht.peekEx(key);
-
- // If near cache does not have value, then we peek DHT cache.
- if (dhtEntry != null) {
- boolean isNew = dhtEntry.isNewLocked() || !dhtEntry.valid(topVer);
-
- if (needVer) {
- T2<CacheObject, GridCacheVersion> res = dhtEntry.innerGetVersioned(
- null,
- /*swap*/true,
- /*unmarshal*/true,
- /**update-metrics*/false,
- /*event*/!isNear && !skipVals,
- subjId,
- null,
- taskName,
- expiryPlc,
- !deserializeBinary);
-
- if (res != null) {
- v = res.get1();
- ver = res.get2();
- }
- }
- else {
- v = dhtEntry.innerGet(tx,
- /*swap*/true,
- /*read-through*/false,
- /*fail-fast*/true,
- /*unmarshal*/true,
- /*update-metrics*/false,
- /*events*/!isNear && !skipVals,
- /*temporary*/false,
- subjId,
- null,
- taskName,
- expiryPlc,
- !deserializeBinary);
- }
-
- // Entry was not in memory or in swap, so we remove it from cache.
- if (v == null && isNew && dhtEntry.markObsoleteIfEmpty(ver))
- dht.removeIfObsolete(key);
- }
-
- if (v != null) {
- if (cctx.cache().configuration().isStatisticsEnabled() && !skipVals)
- near.metrics0().onRead(true);
- }
- else {
- affNode = affinityNode(key, topVer);
-
- if (affNode == null) {
- onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
- "(all partition nodes left the grid)."));
-
- return saved;
- }
-
- if (!affNode.isLocal() && cctx.cache().configuration().isStatisticsEnabled() && !skipVals)
- near.metrics0().onRead(false);
- }
- }
- catch (GridDhtInvalidPartitionException | GridCacheEntryRemovedException ignored) {
- // No-op.
- }
- finally {
- if (dhtEntry != null && (tx == null || (!tx.implicit() && tx.isolation() == READ_COMMITTED))) {
- dht.context().evicts().touch(dhtEntry, topVer);
-
- entry = null;
- }
- }
- }
-
- if (v != null) {
- if (needVer) {
- V val0 = (V)new T2<>(skipVals ? true : v, ver);
+ if (v == null) {
+ boolean fastLocGet = allowLocRead && cctx.allowFastLocalRead(part, affNodes, topVer);
- add(new GridFinishedFuture<>(Collections.singletonMap((K)key, val0)));
- }
- else {
- if (keepCacheObjects) {
- K key0 = (K)key;
- V val0 = (V)(skipVals ? true : v);
+ if (fastLocGet && localDhtGet(key, part, topVer, isNear))
+ break;
- add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
- }
- else {
- K key0 = (K)cctx.unwrapBinaryIfNeeded(key, !deserializeBinary, false);
- V val0 = !skipVals ?
- (V)cctx.unwrapBinaryIfNeeded(v, !deserializeBinary, false) :
- (V)Boolean.TRUE;
+ ClusterNode affNode = affinityNode(affNodes);
- add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
- }
- }
- }
- else {
if (affNode == null) {
- affNode = affinityNode(key, topVer);
-
- if (affNode == null) {
- onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
- "(all partition nodes left the grid)."));
+ onDone(serverNotFoundError(topVer));
- return saved;
- }
+ return saved;
}
+ if (cctx.cache().configuration().isStatisticsEnabled() && !skipVals && !affNode.isLocal())
+ cache().metrics0().onRead(false);
+
LinkedHashMap<KeyCacheObject, Boolean> keys = mapped.get(affNode);
if (keys != null && keys.containsKey(key)) {
@@ -586,7 +496,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
}
- if (!cctx.affinity().localNode(key, topVer)) {
+ if (!affNodes.contains(cctx.localNode())) {
GridNearCacheEntry nearEntry = entry != null ? entry : near.entryExx(key, topVer);
nearEntry.reserveEviction();
@@ -612,6 +522,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
old.put(key, addRdr);
}
+ else
+ addResult(key, v, ver);
break;
}
@@ -633,6 +545,135 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
}
/**
+ * @param key Key.
+ * @param part Partition.
+ * @param topVer Topology version.
+ * @param nearRead {@code True} if already tried to read from near cache.
+ * @return {@code True} if there is no need to further search value.
+ */
+ private boolean localDhtGet(KeyCacheObject key,
+ int part,
+ AffinityTopologyVersion topVer,
+ boolean nearRead) {
+ GridDhtCacheAdapter<K, V> dht = cache().dht();
+
+ assert dht.context().affinityNode() : this;
+
+ while (true) {
+ GridCacheEntryEx dhtEntry = null;
+
+ try {
+ dhtEntry = dht.context().isSwapOrOffheapEnabled() ? dht.entryEx(key) : dht.peekEx(key);
+
+ CacheObject v = null;
+
+ // If near cache does not have value, then we peek DHT cache.
+ if (dhtEntry != null) {
+ boolean isNew = dhtEntry.isNewLocked() || !dhtEntry.valid(topVer);
+
+ if (needVer) {
+ T2<CacheObject, GridCacheVersion> res = dhtEntry.innerGetVersioned(
+ null,
+ /*swap*/true,
+ /*unmarshal*/true,
+ /**update-metrics*/false,
+ /*event*/!nearRead && !skipVals,
+ subjId,
+ null,
+ taskName,
+ expiryPlc,
+ !deserializeBinary);
+
+ if (res != null) {
+ v = res.get1();
+ ver = res.get2();
+ }
+ }
+ else {
+ v = dhtEntry.innerGet(tx,
+ /*swap*/true,
+ /*read-through*/false,
+ /*fail-fast*/true,
+ /*unmarshal*/true,
+ /*update-metrics*/false,
+ /*events*/!nearRead && !skipVals,
+ /*temporary*/false,
+ subjId,
+ null,
+ taskName,
+ expiryPlc,
+ !deserializeBinary);
+ }
+
+ // Entry was not in memory or in swap, so we remove it from cache.
+ if (v == null && isNew && dhtEntry.markObsoleteIfEmpty(ver))
+ dht.removeIfObsolete(key);
+ }
+
+ if (v != null) {
+ if (cctx.cache().configuration().isStatisticsEnabled() && !skipVals)
+ cache().metrics0().onRead(true);
+
+ addResult(key, v, ver);
+
+ return true;
+ }
+ else {
+ boolean topStable = cctx.isReplicated() || topVer.equals(cctx.topology().topologyVersion());
+
+ // Entry not found, do not continue search if topology did not change and there is no store.
+ return !cctx.readThroughConfigured() && (topStable || partitionOwned(part));
+ }
+ }
+ catch (GridCacheEntryRemovedException ignored) {
+ // Retry.
+ }
+ catch (GridDhtInvalidPartitionException e) {
+ return false;
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+
+ return false;
+ }
+ finally {
+ if (dhtEntry != null && (tx == null || (!tx.implicit() && tx.isolation() == READ_COMMITTED)))
+ dht.context().evicts().touch(dhtEntry, topVer);
+ }
+ }
+ }
+
+ /**
+ * @param key Key.
+ * @param v Value.
+ * @param ver Version.
+ */
+ @SuppressWarnings("unchecked")
+ private void addResult(KeyCacheObject key, CacheObject v, GridCacheVersion ver) {
+ if (needVer) {
+ V val0 = (V)new T2<>(skipVals ? true : v, ver);
+
+ add(new GridFinishedFuture<>(Collections.singletonMap((K)key, val0)));
+ }
+ else {
+ if (keepCacheObjects) {
+ K key0 = (K)key;
+ V val0 = (V)(skipVals ? true : v);
+
+ add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
+ }
+ else {
+ K key0 = (K)cctx.unwrapBinaryIfNeeded(key, !deserializeBinary, false);
+ V val0 = !skipVals ?
+ (V)cctx.unwrapBinaryIfNeeded(v, !deserializeBinary, false) :
+ (V)Boolean.TRUE;
+
+ add(new GridFinishedFuture<>(Collections.singletonMap(key0, val0)));
+ }
+ }
+ }
+
+ /**
* @return Near cache.
*/
private GridNearCacheAdapter<K, V> cache() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index ca15e20..7a3b8ff 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -619,17 +619,19 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
return topVer;
}
- for (GridCacheContext cacheCtx : cctx.cache().context().cacheContexts()) {
- if (!cacheCtx.systemTx())
- continue;
+ if (!sysThreadMap.isEmpty()) {
+ for (GridCacheContext cacheCtx : cctx.cache().context().cacheContexts()) {
+ if (!cacheCtx.systemTx())
+ continue;
- tx = sysThreadMap.get(new TxThreadKey(threadId, cacheCtx.cacheId()));
+ tx = sysThreadMap.get(new TxThreadKey(threadId, cacheCtx.cacheId()));
- if (tx != null && tx != ignore) {
- AffinityTopologyVersion topVer = tx.topologyVersionSnapshot();
+ if (tx != null && tx != ignore) {
+ AffinityTopologyVersion topVer = tx.topologyVersionSnapshot();
- if (topVer != null)
- return topVer;
+ if (topVer != null)
+ return topVer;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
new file mode 100644
index 0000000..8a602ad
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
@@ -0,0 +1,157 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_GRID_NAME;
+
+/**
+ *
+ */
+public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
+ /** */
+ private Class<?> recordCls;
+
+ /** */
+ private List<Object> recordedMsgs = new ArrayList<>();
+
+ /** */
+ private List<T2<ClusterNode, GridIoMessage>> blockedMsgs = new ArrayList<>();
+
+ /** */
+ private Map<Class<?>, Set<String>> blockCls = new HashMap<>();
+
+ /** */
+ private IgnitePredicate<GridIoMessage> blockP;
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
+ throws IgniteSpiException {
+ if (msg instanceof GridIoMessage) {
+ GridIoMessage ioMsg = (GridIoMessage)msg;
+
+ Object msg0 = ioMsg.message();
+
+ synchronized (this) {
+ if (recordCls != null && msg0.getClass().equals(recordCls))
+ recordedMsgs.add(msg0);
+
+ boolean block = false;
+
+ if (blockP != null && blockP.apply(ioMsg))
+ block = true;
+ else {
+ Set<String> blockNodes = blockCls.get(msg0.getClass());
+
+ if (blockNodes != null) {
+ String nodeName = (String)node.attributes().get(ATTR_GRID_NAME);
+
+ block = blockNodes.contains(nodeName);
+ }
+ }
+
+ if (block) {
+ blockedMsgs.add(new T2<>(node, ioMsg));
+
+ return;
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+
+ /**
+ * @param recordCls Message class to record.
+ */
+ public void record(@Nullable Class<?> recordCls) {
+ synchronized (this) {
+ this.recordCls = recordCls;
+ }
+ }
+
+ /**
+ * @return Recorded messages.
+ */
+ public List<Object> recordedMessages() {
+ synchronized (this) {
+ List<Object> msgs = recordedMsgs;
+
+ recordedMsgs = new ArrayList<>();
+
+ return msgs;
+ }
+ }
+
+ /**
+ * @param blockP Message block predicate.
+ */
+ public void blockMessages(IgnitePredicate<GridIoMessage> blockP) {
+ synchronized (this) {
+ this.blockP = blockP;
+ }
+ }
+
+ /**
+ * @param cls Message class.
+ * @param nodeName Node name.
+ */
+ public void blockMessages(Class<?> cls, String nodeName) {
+ synchronized (this) {
+ Set<String> set = blockCls.get(cls);
+
+ if (set == null) {
+ set = new HashSet<>();
+
+ blockCls.put(cls, set);
+ }
+
+ set.add(nodeName);
+ }
+ }
+
+ /**
+ * Stops block messages and sends all already blocked messages.
+ */
+ public void stopBlock() {
+ synchronized (this) {
+ blockCls.clear();
+
+ for (T2<ClusterNode, GridIoMessage> msg : blockedMsgs)
+ super.sendMessage(msg.get1(), msg.get2());
+
+ blockedMsgs.clear();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
index e28e89f..a1f917f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
@@ -19,25 +19,19 @@ package org.apache.ignite.internal.processors.cache;
import java.io.Externalizable;
import java.io.Serializable;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Map;
import java.util.concurrent.Callable;
import javax.cache.Cache;
-import javax.cache.integration.CacheLoaderException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheInterceptorAdapter;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.AffinityFunction;
import org.apache.ignite.cache.affinity.AffinityNodeIdHashResolver;
-import org.apache.ignite.cache.affinity.fair.FairAffinityFunction;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cache.eviction.EvictionFilter;
import org.apache.ignite.cache.eviction.fifo.FifoEvictionPolicy;
import org.apache.ignite.cache.eviction.lru.LruEvictionPolicy;
import org.apache.ignite.cache.eviction.random.RandomEvictionPolicy;
-import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DeploymentMode;
@@ -46,7 +40,6 @@ import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
@@ -54,7 +47,6 @@ import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.GridStringLogger;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
@@ -862,49 +854,9 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
}, IgniteCheckedException.class, null);
}
- /** */
- private static class TestStore implements CacheStore<Object,Object> {
- /** {@inheritDoc} */
- @Nullable @Override public Object load(Object key) {
- return null;
- }
-
- /** {@inheritDoc} */
- @Override public void loadCache(IgniteBiInClosure<Object, Object> clo, @Nullable Object... args) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public Map<Object, Object> loadAll(Iterable<?> keys) throws CacheLoaderException {
- return Collections.emptyMap();
- }
-
- /** {@inheritDoc} */
- @Override public void write(Cache.Entry<?, ?> entry) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void writeAll(Collection<Cache.Entry<?, ?>> entries) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void delete(Object key) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void deleteAll(Collection<?> keys) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void sessionEnd(boolean commit) {
- // No-op.
- }
- }
-
+ /**
+ *
+ */
private static class TestRendezvousAffinityFunction extends RendezvousAffinityFunction {
/**
* Empty constructor required by {@link Externalizable}.
@@ -941,6 +893,10 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
// No-op, just different class.
}
+ /**
+ *
+ */
private static class TestCacheDefaultAffinityKeyMapper extends GridCacheDefaultAffinityKeyMapper {
+ // No-op, just different class.
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
index 100acfe..f106fec 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
@@ -18,22 +18,15 @@
package org.apache.ignite.internal.processors.cache;
import java.util.Collection;
-import java.util.concurrent.ConcurrentLinkedDeque;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheAtomicityMode;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteKernal;
-import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.spi.IgniteSpiException;
-import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -71,7 +64,11 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
if (getTestGridName(0).equals(gridName))
cfg.setClientMode(true);
- cfg.setCommunicationSpi(new TestCommunicationSpi());
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
+
+ commSpi.record(GridNearLockRequest.class);
+
+ cfg.setCommunicationSpi(commSpi);
return cfg;
}
@@ -88,18 +85,18 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
cache.put("key1", "val1");
for (int i = 0; i < 3; i++) {
- ((TestCommunicationSpi)ignite(0).configuration().getCommunicationSpi()).clear();
- ((TestCommunicationSpi)ignite(1).configuration().getCommunicationSpi()).clear();
-
try (Transaction tx = ignite(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
cache.get("key1");
tx.commit();
}
- TestCommunicationSpi comm = (TestCommunicationSpi)ignite(0).configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi comm =
+ (TestRecordingCommunicationSpi)ignite(0).configuration().getCommunicationSpi();
+
+ Collection<GridNearLockRequest> reqs = (Collection)comm.recordedMessages();
- assertEquals(1, comm.requests().size());
+ assertEquals(1, reqs.size());
GridCacheAdapter<Object, Object> primary = ((IgniteKernal)grid(1)).internalCache("partitioned");
@@ -107,7 +104,7 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
assertNotNull(dhtEntry);
- GridNearLockRequest req = comm.requests().iterator().next();
+ GridNearLockRequest req = reqs.iterator().next();
assertEquals(dhtEntry.version(), req.dhtVersion(0));
@@ -122,39 +119,4 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
}
}
}
-
- /**
- *
- */
- private static class TestCommunicationSpi extends TcpCommunicationSpi {
- /** */
- private Collection<GridNearLockRequest> reqs = new ConcurrentLinkedDeque<>();
-
- /** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
- throws IgniteSpiException {
- if (msg instanceof GridIoMessage) {
- GridIoMessage ioMsg = (GridIoMessage)msg;
-
- if (ioMsg.message() instanceof GridNearLockRequest)
- reqs.add((GridNearLockRequest)ioMsg.message());
- }
-
- super.sendMessage(node, msg, ackC);
- }
-
- /**
- * @return Collected requests.
- */
- public Collection<GridNearLockRequest> requests() {
- return reqs;
- }
-
- /**
- *
- */
- public void clear() {
- reqs.clear();
- }
- }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java
index 57d57ca..48acdfc 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java
@@ -22,29 +22,41 @@ import java.util.Collections;
import java.util.Map;
import java.util.Set;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
/**
*
*/
public class IgniteCacheStoreCollectionTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
CacheConfiguration<Object, Object> ccfg1 = new CacheConfiguration<>();
ccfg1.setName("cache1");
ccfg1.setAtomicityMode(ATOMIC);
+ ccfg1.setWriteSynchronizationMode(FULL_SYNC);
CacheConfiguration<Object, Object> ccfg2 = new CacheConfiguration<>();
ccfg2.setName("cache2");
ccfg2.setAtomicityMode(TRANSACTIONAL);
+ ccfg2.setWriteSynchronizationMode(FULL_SYNC);
cfg.setCacheConfiguration(ccfg1, ccfg2);
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
index 9acc4b5..ac80d69 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
@@ -46,6 +46,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
/**
*
@@ -344,6 +345,7 @@ public class IgniteDynamicCacheStartNoExchangeTimeoutTest extends GridCommonAbst
ccfg.setName("cache-1");
ccfg.setAtomicityMode(ATOMIC);
ccfg.setBackups(0);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
res.add(ccfg);
}
@@ -354,6 +356,7 @@ public class IgniteDynamicCacheStartNoExchangeTimeoutTest extends GridCommonAbst
ccfg.setName("cache-2");
ccfg.setAtomicityMode(ATOMIC);
ccfg.setBackups(1);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
res.add(ccfg);
}
@@ -365,6 +368,7 @@ public class IgniteDynamicCacheStartNoExchangeTimeoutTest extends GridCommonAbst
ccfg.setAtomicityMode(ATOMIC);
ccfg.setBackups(1);
ccfg.setAffinity(new FairAffinityFunction());
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
res.add(ccfg);
}
@@ -375,6 +379,7 @@ public class IgniteDynamicCacheStartNoExchangeTimeoutTest extends GridCommonAbst
ccfg.setName("cache-4");
ccfg.setAtomicityMode(TRANSACTIONAL);
ccfg.setBackups(0);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
res.add(ccfg);
}
@@ -385,6 +390,7 @@ public class IgniteDynamicCacheStartNoExchangeTimeoutTest extends GridCommonAbst
ccfg.setName("cache-5");
ccfg.setAtomicityMode(TRANSACTIONAL);
ccfg.setBackups(1);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
res.add(ccfg);
}
@@ -396,6 +402,7 @@ public class IgniteDynamicCacheStartNoExchangeTimeoutTest extends GridCommonAbst
ccfg.setAtomicityMode(TRANSACTIONAL);
ccfg.setBackups(1);
ccfg.setAffinity(new FairAffinityFunction());
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
res.add(ccfg);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePartitionNotLoadedEventSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePartitionNotLoadedEventSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePartitionNotLoadedEventSelfTest.java
index 5bc779c..6a42752 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePartitionNotLoadedEventSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePartitionNotLoadedEventSelfTest.java
@@ -22,7 +22,6 @@ import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -42,6 +41,9 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.util.TestTcpCommunicationSpi;
import org.eclipse.jetty.util.ConcurrentHashSet;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
/**
*
*/
@@ -76,8 +78,9 @@ public class GridCachePartitionNotLoadedEventSelfTest extends GridCommonAbstract
CacheConfiguration<Integer, Integer> cacheCfg = new CacheConfiguration<>();
- cacheCfg.setCacheMode(CacheMode.PARTITIONED);
+ cacheCfg.setCacheMode(PARTITIONED);
cacheCfg.setBackups(backupCnt);
+ cacheCfg.setWriteSynchronizationMode(FULL_SYNC);
cfg.setCacheConfiguration(cacheCfg);
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeRestartTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeRestartTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeRestartTest.java
index 327db0e..37ed866 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeRestartTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeRestartTest.java
@@ -31,10 +31,12 @@ public class IgniteCacheAtomicNodeRestartTest extends GridCachePartitionedNodeRe
return ATOMIC;
}
+ /** {@inheritDoc} */
@Override public void testRestartWithPutFourNodesNoBackups() {
fail("https://issues.apache.org/jira/browse/IGNITE-1587");
}
+ /** {@inheritDoc} */
@Override public void testRestartWithPutFourNodesOneBackupsOffheapTiered() {
fail("https://issues.apache.org/jira/browse/IGNITE-1587");
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/83b2bf5e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
index e7657a6..13f2598 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
@@ -2010,7 +2010,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
private List<Object> recordedMsgs = new ArrayList<>();
/** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackClosure)
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
throws IgniteSpiException {
if (msg instanceof GridIoMessage) {
Object msg0 = ((GridIoMessage)msg).message();
@@ -2032,7 +2032,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
}
}
- super.sendMessage(node, msg, ackClosure);
+ super.sendMessage(node, msg, ackC);
}
/**
[08/28] ignite git commit: Merge remote-tracking branch
'origin/master'
Posted by vo...@apache.org.
Merge remote-tracking branch 'origin/master'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/add8379c
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/add8379c
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/add8379c
Branch: refs/heads/ignite-2236
Commit: add8379cf037dbb0c0b7fd1c337abdb429d95358
Parents: 422272d 9d139e5
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Mon Jan 18 12:32:58 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 12:32:58 2016 +0300
----------------------------------------------------------------------
.../IgniteServiceDynamicCachesSelfTest.java | 183 +++++++++++++++++++
1 file changed, 183 insertions(+)
----------------------------------------------------------------------
[24/28] ignite git commit: IgniteCacheP2pUnmarshallingQueryErrorTest
test unmuted
Posted by vo...@apache.org.
IgniteCacheP2pUnmarshallingQueryErrorTest test unmuted
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/67791457
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/67791457
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/67791457
Branch: refs/heads/ignite-2236
Commit: 67791457f87097469f0efadcea6c6d13dd2ed37f
Parents: 7ca1a8d
Author: agura <ag...@gridgain.com>
Authored: Mon Jan 18 21:36:21 2016 +0300
Committer: agura <ag...@gridgain.com>
Committed: Mon Jan 18 21:37:13 2016 +0300
----------------------------------------------------------------------
.../cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/67791457/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
index cb7e3ad..c9544b9 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
@@ -63,8 +63,6 @@ public class IgniteCacheP2pUnmarshallingQueryErrorTest extends IgniteCacheP2pUnm
* @throws Exception If failed.
*/
public void testResponseMessageOnRequestUnmarshallingFailed() throws Exception {
- fail("https://issues.apache.org/jira/browse/IGNITE-1923");
-
readCnt.set(Integer.MAX_VALUE);
try {
[20/28] ignite git commit: Merge remote-tracking branch
'origin/master'
Posted by vo...@apache.org.
Merge remote-tracking branch 'origin/master'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d7fd5807
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d7fd5807
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d7fd5807
Branch: refs/heads/ignite-2236
Commit: d7fd5807763330c80536e5fe4b9cd8e8129f0a57
Parents: 22cf3a3 83b2bf5
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Mon Jan 18 18:33:00 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 18:33:00 2016 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheAdapter.java | 30 +-
.../processors/cache/GridCacheContext.java | 33 ++
.../dht/CacheDistributedGetFutureAdapter.java | 28 +-
.../dht/GridClientPartitionTopology.java | 2 +
.../dht/GridDhtPartitionTopologyImpl.java | 27 +-
.../dht/GridPartitionedGetFuture.java | 241 ++++++-----
.../dht/GridPartitionedSingleGetFuture.java | 229 ++++++----
.../dht/atomic/GridDhtAtomicCache.java | 26 ++
.../colocated/GridDhtColocatedLockFuture.java | 4 -
.../distributed/near/GridNearGetFuture.java | 267 +++++++-----
.../cache/transactions/IgniteTxManager.java | 18 +-
.../internal/TestRecordingCommunicationSpi.java | 157 +++++++
...idCacheConfigurationConsistencySelfTest.java | 58 +--
.../cache/IgniteCacheNearLockValueSelfTest.java | 62 +--
.../cache/IgniteCacheStoreCollectionTest.java | 12 +
...eDynamicCacheStartNoExchangeTimeoutTest.java | 7 +
...ridCachePartitionNotLoadedEventSelfTest.java | 7 +-
.../IgniteCacheAtomicNodeRestartTest.java | 2 +
...niteCacheClientNodeChangingTopologyTest.java | 4 +-
.../distributed/IgniteCacheGetRestartTest.java | 280 ++++++++++++
.../IgniteCacheReadFromBackupTest.java | 427 +++++++++++++++++++
.../IgniteCacheSingleGetMessageTest.java | 88 +---
.../IgniteCrossCacheTxStoreSelfTest.java | 1 +
.../GridCacheDhtPreloadMessageCountTest.java | 62 +--
.../near/GridCacheGetStoreErrorSelfTest.java | 9 +-
.../GridCachePartitionedNodeRestartTest.java | 4 +-
...ePartitionedOptimisticTxNodeRestartTest.java | 4 +-
.../IgniteCacheRestartTestSuite2.java | 3 +
.../testsuites/IgniteCacheTestSuite4.java | 2 +
...niteCacheP2pUnmarshallingQueryErrorTest.java | 18 +-
30 files changed, 1527 insertions(+), 585 deletions(-)
----------------------------------------------------------------------
[25/28] ignite git commit: Merge branch 'master' into ignite-2236
Posted by vo...@apache.org.
Merge branch 'master' into ignite-2236
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/fd574b6a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/fd574b6a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/fd574b6a
Branch: refs/heads/ignite-2236
Commit: fd574b6ac90c0828c62a0552e906d722df3bcc85
Parents: 7d8e304 6779145
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Jan 19 11:26:43 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Jan 19 11:26:43 2016 +0300
----------------------------------------------------------------------
.../internal/GridEventConsumeHandler.java | 3 +-
.../internal/GridMessageListenHandler.java | 6 +-
.../internal/binary/BinaryClassDescriptor.java | 15 +-
.../internal/binary/BinaryEnumObjectImpl.java | 3 +-
.../managers/discovery/CustomEventListener.java | 4 +-
.../discovery/GridDiscoveryManager.java | 8 +-
.../processors/cache/GridCacheAdapter.java | 30 +-
.../processors/cache/GridCacheContext.java | 33 ++
.../dht/CacheDistributedGetFutureAdapter.java | 28 +-
.../dht/GridClientPartitionTopology.java | 2 +
.../dht/GridDhtPartitionTopologyImpl.java | 27 +-
.../dht/GridPartitionedGetFuture.java | 241 ++++++-----
.../dht/GridPartitionedSingleGetFuture.java | 229 ++++++----
.../dht/atomic/GridDhtAtomicCache.java | 26 ++
.../colocated/GridDhtColocatedLockFuture.java | 4 -
.../distributed/near/GridNearGetFuture.java | 267 +++++++-----
.../continuous/CacheContinuousQueryHandler.java | 36 +-
.../cache/transactions/IgniteTxManager.java | 18 +-
.../continuous/GridContinuousHandler.java | 4 +-
.../continuous/GridContinuousProcessor.java | 15 +-
.../internal/processors/igfs/IgfsProcessor.java | 3 -
.../platform/PlatformAbstractTarget.java | 28 +-
.../platform/PlatformNoopProcessor.java | 5 +
.../processors/platform/PlatformProcessor.java | 8 +
.../platform/PlatformProcessorImpl.java | 5 +
.../platform/cache/PlatformCache.java | 6 +-
.../platform/compute/PlatformCompute.java | 132 +++++-
.../dotnet/PlatformDotNetCacheStore.java | 33 +-
.../platform/events/PlatformEvents.java | 16 +-
.../platform/messaging/PlatformMessaging.java | 7 +-
.../platform/services/PlatformServices.java | 14 +-
.../ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java | 48 ++-
.../internal/TestRecordingCommunicationSpi.java | 157 +++++++
...idCacheConfigurationConsistencySelfTest.java | 58 +--
.../cache/IgniteCacheNearLockValueSelfTest.java | 62 +--
.../cache/IgniteCacheStoreCollectionTest.java | 12 +
...eDynamicCacheStartNoExchangeTimeoutTest.java | 7 +
...ridCachePartitionNotLoadedEventSelfTest.java | 7 +-
.../IgniteCacheAtomicNodeRestartTest.java | 2 +
...niteCacheClientNodeChangingTopologyTest.java | 4 +-
.../distributed/IgniteCacheGetRestartTest.java | 280 ++++++++++++
.../IgniteCacheReadFromBackupTest.java | 427 +++++++++++++++++++
.../IgniteCacheSingleGetMessageTest.java | 88 +---
.../IgniteCrossCacheTxStoreSelfTest.java | 1 +
.../GridCacheDhtPreloadMessageCountTest.java | 62 +--
.../near/GridCacheGetStoreErrorSelfTest.java | 9 +-
.../GridCachePartitionedNodeRestartTest.java | 4 +-
...ePartitionedOptimisticTxNodeRestartTest.java | 4 +-
...ContinuousQueryFailoverAbstractSelfTest.java | 80 ++++
.../CacheContinuousQueryLostPartitionTest.java | 256 +++++++++++
.../IgniteServiceDynamicCachesSelfTest.java | 183 ++++++++
.../platform/PlatformComputeBroadcastTask.java | 7 +
...CommunicationRecoveryAckClosureSelfTest.java | 2 +
.../IgniteCacheRestartTestSuite2.java | 3 +
.../testsuites/IgniteCacheTestSuite4.java | 2 +
.../hadoop/fs/BasicHadoopFileSystemFactory.java | 19 +-
.../fs/IgniteHadoopIgfsSecondaryFileSystem.java | 6 +-
.../hadoop/fs/v1/IgniteHadoopFileSystem.java | 7 +-
.../hadoop/fs/v2/IgniteHadoopFileSystem.java | 7 +
...niteCacheP2pUnmarshallingQueryErrorTest.java | 18 +-
.../IgniteBinaryCacheQueryTestSuite.java | 2 +
.../cpp/common/include/ignite/common/exports.h | 1 +
.../cpp/common/include/ignite/common/java.h | 3 +
.../platforms/cpp/common/project/vs/module.def | 3 +-
modules/platforms/cpp/common/src/exports.cpp | 4 +
modules/platforms/cpp/common/src/java.cpp | 20 +
.../Apache.Ignite.Core.Tests.csproj | 1 +
.../Binary/BinaryBuilderSelfTest.cs | 8 +-
.../Cache/CacheAbstractTest.cs | 19 +
.../Compute/CancellationTest.cs | 174 ++++++++
.../Compute/ComputeApiTest.cs | 2 +-
.../Apache.Ignite.Core/Binary/IBinaryObject.cs | 8 +
.../Common/IgniteFutureCancelledException.cs | 1 +
.../dotnet/Apache.Ignite.Core/IIgnite.cs | 7 +
.../Apache.Ignite.Core/Impl/Binary/Binary.cs | 3 +
.../Impl/Binary/BinaryEnum.cs | 6 +
.../Impl/Binary/BinaryObject.cs | 6 +
.../Impl/Cache/Store/CacheStore.cs | 9 +-
.../dotnet/Apache.Ignite.Core/Impl/Ignite.cs | 6 +
.../Apache.Ignite.Core/Impl/IgniteProxy.cs | 6 +
.../Impl/Unmanaged/IgniteJniNativeMethods.cs | 3 +
.../Impl/Unmanaged/UnmanagedUtils.cs | 22 +-
.../Services/ServicesExample.cs | 2 +-
83 files changed, 2661 insertions(+), 733 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/fd574b6a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/CacheDistributedGetFutureAdapter.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/fd574b6a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
----------------------------------------------------------------------
[05/28] ignite git commit: IGNITE-2376: .NET: Fixed interop processor
leak. This closes #402.
Posted by vo...@apache.org.
IGNITE-2376: .NET: Fixed interop processor leak. This closes #402.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/26251290
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/26251290
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/26251290
Branch: refs/heads/ignite-2236
Commit: 2625129057f116dd037089a7078f1dab4fbb6b2e
Parents: b20dc40
Author: Pavel Tupitsyn <pt...@gridgain.com>
Authored: Mon Jan 18 12:21:40 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 12:21:40 2016 +0300
----------------------------------------------------------------------
.../dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/26251290/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
index 4c8f1dc..e9800ee 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
@@ -57,7 +57,7 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
#region NATIVE METHODS: PROCESSOR
- internal static IUnmanagedTarget IgnitionStart(UnmanagedContext ctx, string cfgPath, string gridName,
+ internal static void IgnitionStart(UnmanagedContext ctx, string cfgPath, string gridName,
bool clientMode)
{
using (var mem = IgniteManager.Memory.Allocate().GetStream())
@@ -69,10 +69,12 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
try
{
+ // OnStart receives the same InteropProcessor as here (just as another GlobalRef) and stores it.
+ // Release current reference immediately.
void* res = JNI.IgnitionStart(ctx.NativeContext, cfgPath0, gridName0, InteropFactoryId,
mem.SynchronizeOutput());
- return new UnmanagedTarget(ctx, res);
+ JNI.Release(res);
}
finally
{
[10/28] ignite git commit: IGNITE-2228: Platform compute futures
cancellation support. This closes #394.
Posted by vo...@apache.org.
IGNITE-2228: Platform compute futures cancellation support. This closes #394.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/da601c27
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/da601c27
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/da601c27
Branch: refs/heads/ignite-2236
Commit: da601c27de1a875831acbcba9b563ed44df7bc98
Parents: 2e1d50d
Author: Pavel Tupitsyn <pt...@gridgain.com>
Authored: Mon Jan 18 13:01:43 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 13:01:43 2016 +0300
----------------------------------------------------------------------
.../platform/PlatformAbstractTarget.java | 28 +--
.../platform/cache/PlatformCache.java | 6 +-
.../platform/compute/PlatformCompute.java | 132 ++++++++++++--
.../platform/events/PlatformEvents.java | 16 +-
.../platform/messaging/PlatformMessaging.java | 7 +-
.../platform/services/PlatformServices.java | 14 +-
.../platform/PlatformComputeBroadcastTask.java | 7 +
.../Apache.Ignite.Core.Tests.csproj | 1 +
.../Compute/CancellationTest.cs | 174 +++++++++++++++++++
.../Compute/ComputeApiTest.cs | 2 +-
10 files changed, 329 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
index 7ffceef..0cd683d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
@@ -24,9 +24,8 @@ import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.memory.PlatformMemory;
import org.apache.ignite.internal.processors.platform.memory.PlatformOutputStream;
-import org.apache.ignite.internal.processors.platform.utils.*;
-import org.apache.ignite.internal.util.future.IgniteFutureImpl;
-import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.internal.processors.platform.utils.PlatformFutureUtils;
+import org.apache.ignite.internal.processors.platform.utils.PlatformListenable;
import org.jetbrains.annotations.Nullable;
/**
@@ -194,26 +193,13 @@ public abstract class PlatformAbstractTarget implements PlatformTarget {
/** {@inheritDoc} */
@Override public PlatformListenable listenFutureAndGet(final long futId, int typ) throws Exception {
- return PlatformFutureUtils.listen(platformCtx, currentFutureWrapped(), futId, typ, null, this);
+ return PlatformFutureUtils.listen(platformCtx, currentFuture(), futId, typ, null, this);
}
/** {@inheritDoc} */
@Override public PlatformListenable listenFutureForOperationAndGet(final long futId, int typ, int opId)
throws Exception {
- return PlatformFutureUtils.listen(platformCtx, currentFutureWrapped(), futId, typ, futureWriter(opId), this);
- }
-
- /**
- * Get current future with proper exception conversions.
- *
- * @return Future.
- * @throws IgniteCheckedException If failed.
- */
- @SuppressWarnings({"ThrowableResultOfMethodCallIgnored", "unchecked"})
- protected IgniteInternalFuture currentFutureWrapped() throws IgniteCheckedException {
- IgniteFutureImpl fut = (IgniteFutureImpl)currentFuture();
-
- return fut.internalFuture();
+ return PlatformFutureUtils.listen(platformCtx, currentFuture(), futId, typ, futureWriter(opId), this);
}
/**
@@ -222,8 +208,8 @@ public abstract class PlatformAbstractTarget implements PlatformTarget {
* @return current future.
* @throws IgniteCheckedException
*/
- protected IgniteFuture currentFuture() throws IgniteCheckedException {
- throw new IgniteCheckedException("Future listening is not supported in " + this.getClass());
+ protected IgniteInternalFuture currentFuture() throws IgniteCheckedException {
+ throw new IgniteCheckedException("Future listening is not supported in " + getClass());
}
/**
@@ -232,7 +218,7 @@ public abstract class PlatformAbstractTarget implements PlatformTarget {
* @param opId Operation id.
* @return A custom writer for given op id.
*/
- protected @Nullable PlatformFutureUtils.Writer futureWriter(int opId){
+ @Nullable protected PlatformFutureUtils.Writer futureWriter(int opId){
return null;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCache.java
index 2f7cab2..8e7c51d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCache.java
@@ -28,6 +28,7 @@ import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.cache.query.TextQuery;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.cache.CacheOperationContext;
@@ -43,6 +44,7 @@ import org.apache.ignite.internal.processors.platform.cache.query.PlatformQueryC
import org.apache.ignite.internal.processors.platform.utils.PlatformFutureUtils;
import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
import org.apache.ignite.internal.util.GridConcurrentFactory;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.lang.IgniteFuture;
import org.jetbrains.annotations.Nullable;
@@ -684,8 +686,8 @@ public class PlatformCache extends PlatformAbstractTarget {
}
/** <inheritDoc /> */
- @Override protected IgniteFuture currentFuture() throws IgniteCheckedException {
- return cache.future();
+ @Override protected IgniteInternalFuture currentFuture() throws IgniteCheckedException {
+ return ((IgniteFutureImpl)cache.future()).internalFuture();
}
/** <inheritDoc /> */
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
index 1dad126..10545d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
@@ -17,12 +17,10 @@
package org.apache.ignite.internal.processors.platform.compute;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.compute.ComputeTaskFuture;
import org.apache.ignite.internal.IgniteComputeImpl;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryObjectImpl;
@@ -30,11 +28,17 @@ import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.PlatformAbstractTarget;
import org.apache.ignite.internal.processors.platform.PlatformContext;
-import org.apache.ignite.internal.processors.platform.utils.*;
-import org.apache.ignite.internal.util.typedef.C1;
-import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.internal.processors.platform.utils.PlatformFutureUtils;
+import org.apache.ignite.internal.processors.platform.utils.PlatformListenable;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
+import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.binary.BinaryObject;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBGRID;
@@ -62,7 +66,7 @@ public class PlatformCompute extends PlatformAbstractTarget {
private final IgniteComputeImpl compute;
/** Future for previous asynchronous operation. */
- protected ThreadLocal<IgniteFuture<?>> curFut = new ThreadLocal<>();
+ protected ThreadLocal<IgniteInternalFuture> curFut = new ThreadLocal<>();
/**
* Constructor.
*
@@ -213,8 +217,8 @@ public class PlatformCompute extends PlatformAbstractTarget {
}
/** <inheritDoc /> */
- @Override protected IgniteFuture currentFuture() throws IgniteCheckedException {
- IgniteFuture<?> fut = curFut.get();
+ @Override protected IgniteInternalFuture currentFuture() throws IgniteCheckedException {
+ IgniteInternalFuture fut = curFut.get();
if (fut == null)
throw new IllegalStateException("Asynchronous operation not started.");
@@ -272,13 +276,7 @@ public class PlatformCompute extends PlatformAbstractTarget {
Object res = compute0.execute(taskName, arg);
if (async) {
- curFut.set(compute0.future().chain(new C1<IgniteFuture, Object>() {
- private static final long serialVersionUID = 0L;
-
- @Override public Object apply(IgniteFuture fut) {
- return toBinary(fut.get());
- }
- }));
+ curFut.set(new ComputeConvertingFuture(compute0.future()));
return null;
}
@@ -327,4 +325,102 @@ public class PlatformCompute extends PlatformAbstractTarget {
return nodeIds == null ? compute :
platformCtx.kernalContext().grid().compute(compute.clusterGroup().forNodeIds(nodeIds));
}
+
+ /**
+ * Wraps ComputeTaskFuture as IgniteInternalFuture.
+ */
+ protected class ComputeConvertingFuture implements IgniteInternalFuture {
+ /** */
+ private final IgniteInternalFuture fut;
+
+ /**
+ * Ctor.
+ *
+ * @param fut Future to wrap.
+ */
+ public ComputeConvertingFuture(ComputeTaskFuture fut) {
+ this.fut = ((IgniteFutureImpl)fut).internalFuture();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object get() throws IgniteCheckedException {
+ return convertResult(fut.get());
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object get(long timeout) throws IgniteCheckedException {
+ return convertResult(fut.get(timeout));
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object get(long timeout, TimeUnit unit) throws IgniteCheckedException {
+ return convertResult(fut.get(timeout, unit));
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object getUninterruptibly() throws IgniteCheckedException {
+ return convertResult(fut.get());
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean cancel() throws IgniteCheckedException {
+ return fut.cancel();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isDone() {
+ return fut.isDone();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isCancelled() {
+ return fut.isCancelled();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long startTime() {
+ return fut.startTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long duration() {
+ return fut.duration();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void listen(final IgniteInClosure lsnr) {
+ fut.listen(new IgniteInClosure<IgniteInternalFuture>() {
+ private static final long serialVersionUID = 0L;
+
+ @Override public void apply(IgniteInternalFuture fut0) {
+ lsnr.apply(ComputeConvertingFuture.this);
+ }
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture chain(IgniteClosure doneCb) {
+ throw new UnsupportedOperationException("Chain operation is not supported.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public Throwable error() {
+ return fut.error();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object result() {
+ return convertResult(fut.result());
+ }
+
+ /**
+ * Converts future result.
+ *
+ * @param obj Object to convert.
+ * @return Result.
+ */
+ protected Object convertResult(Object obj) {
+ return toBinary(obj);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/events/PlatformEvents.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/events/PlatformEvents.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/events/PlatformEvents.java
index 9bf0a8d..71708af 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/events/PlatformEvents.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/events/PlatformEvents.java
@@ -17,24 +17,26 @@
package org.apache.ignite.internal.processors.platform.events;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteEvents;
import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventAdapter;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.PlatformAbstractTarget;
-import org.apache.ignite.internal.processors.platform.PlatformEventFilterListener;
import org.apache.ignite.internal.processors.platform.PlatformContext;
+import org.apache.ignite.internal.processors.platform.PlatformEventFilterListener;
import org.apache.ignite.internal.processors.platform.utils.PlatformFutureUtils;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgnitePredicate;
import org.jetbrains.annotations.Nullable;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.UUID;
+
/**
* Interop events.
*/
@@ -269,8 +271,8 @@ public class PlatformEvents extends PlatformAbstractTarget {
}
/** <inheritDoc /> */
- @Override protected IgniteFuture currentFuture() throws IgniteCheckedException {
- return events.future();
+ @Override protected IgniteInternalFuture currentFuture() throws IgniteCheckedException {
+ return ((IgniteFutureImpl)events.future()).internalFuture();
}
/** <inheritDoc /> */
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/messaging/PlatformMessaging.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/messaging/PlatformMessaging.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/messaging/PlatformMessaging.java
index 88ea3c8..619fea7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/messaging/PlatformMessaging.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/messaging/PlatformMessaging.java
@@ -19,13 +19,14 @@ package org.apache.ignite.internal.processors.platform.messaging;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteMessaging;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.PlatformAbstractTarget;
import org.apache.ignite.internal.processors.platform.PlatformContext;
import org.apache.ignite.internal.processors.platform.message.PlatformMessageFilter;
import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
-import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import java.util.UUID;
@@ -160,7 +161,7 @@ public class PlatformMessaging extends PlatformAbstractTarget {
}
/** <inheritDoc /> */
- @Override protected IgniteFuture currentFuture() throws IgniteCheckedException {
- return messaging.future();
+ @Override protected IgniteInternalFuture currentFuture() throws IgniteCheckedException {
+ return ((IgniteFutureImpl)messaging.future()).internalFuture();
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformServices.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformServices.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformServices.java
index 9676b6f..963c72e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformServices.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/services/PlatformServices.java
@@ -17,11 +17,9 @@
package org.apache.ignite.internal.processors.platform.services;
-import java.util.Collection;
-import java.util.Map;
-import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteServices;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.PlatformAbstractTarget;
@@ -31,12 +29,16 @@ import org.apache.ignite.internal.processors.platform.dotnet.PlatformDotNetServi
import org.apache.ignite.internal.processors.platform.utils.PlatformUtils;
import org.apache.ignite.internal.processors.platform.utils.PlatformWriterBiClosure;
import org.apache.ignite.internal.processors.platform.utils.PlatformWriterClosure;
-import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.services.Service;
import org.apache.ignite.services.ServiceConfiguration;
import org.apache.ignite.services.ServiceDescriptor;
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+
/**
* Interop services.
*/
@@ -269,7 +271,7 @@ public class PlatformServices extends PlatformAbstractTarget {
}
/** <inheritDoc /> */
- @Override protected IgniteFuture currentFuture() throws IgniteCheckedException {
- return services.future();
+ @Override protected IgniteInternalFuture currentFuture() throws IgniteCheckedException {
+ return ((IgniteFutureImpl)services.future()).internalFuture();
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeBroadcastTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeBroadcastTask.java b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeBroadcastTask.java
index c721e16..7bcba33 100644
--- a/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeBroadcastTask.java
+++ b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeBroadcastTask.java
@@ -67,6 +67,13 @@ public class PlatformComputeBroadcastTask extends ComputeTaskAdapter<Object, Col
/** {@inheritDoc} */
@Nullable @Override public Object execute() {
+ try {
+ Thread.sleep(50); // Short sleep for cancellation tests.
+ }
+ catch (InterruptedException ignored) {
+ // No-op.
+ }
+
return ignite.cluster().localNode().id();
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
index 72c0210..a247f63 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
@@ -86,6 +86,7 @@
<Compile Include="Cache\Store\CacheStoreTest.cs" />
<Compile Include="Cache\Store\CacheTestParallelLoadStore.cs" />
<Compile Include="Cache\Store\CacheTestStore.cs" />
+ <Compile Include="Compute\CancellationTest.cs" />
<Compile Include="Compute\Forked\ForkedBinarizableClosureTaskTest.cs" />
<Compile Include="Compute\Forked\ForkedResourceTaskTest.cs" />
<Compile Include="Compute\Forked\ForkedSerializableClosureTaskTest.cs" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/CancellationTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/CancellationTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/CancellationTest.cs
new file mode 100644
index 0000000..bbd1169
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/CancellationTest.cs
@@ -0,0 +1,174 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Core.Tests.Compute
+{
+ using System;
+ using System.Collections.Generic;
+ using System.Linq;
+ using System.Threading;
+ using Apache.Ignite.Core.Cluster;
+ using Apache.Ignite.Core.Compute;
+ using NUnit.Framework;
+
+ /// <summary>
+ /// Cancellation tests.
+ /// </summary>
+ public class CancellationTest : IgniteTestBase
+ {
+ public CancellationTest()
+ : base("config\\compute\\compute-grid1.xml", "config\\compute\\compute-grid2.xml")
+ {
+ // No-op.
+ }
+
+ [Test]
+ public void TestTask()
+ {
+ TestTask((c, t) => c.ExecuteAsync(new Task(), t));
+ TestTask((c, t) => c.ExecuteAsync(new Task(), 1, t));
+ TestTask((c, t) => c.ExecuteAsync<int, IList<IComputeJobResult<int>>>(typeof(Task), t));
+ TestTask((c, t) => c.ExecuteAsync<object, int, IList<IComputeJobResult<int>>>(typeof(Task), 1, t));
+ }
+
+ [Test]
+ public void TestJavaTask()
+ {
+ using (var cts = new CancellationTokenSource())
+ {
+ var task = Compute.ExecuteJavaTaskAsync<object>(ComputeApiTest.BroadcastTask, null, cts.Token);
+
+ Assert.IsFalse(task.IsCanceled);
+
+ cts.Cancel();
+
+ Assert.IsTrue(task.IsCanceled);
+
+ // Pass cancelled token
+ Assert.IsTrue(
+ Compute.ExecuteJavaTaskAsync<object>(ComputeApiTest.BroadcastTask, null, cts.Token).IsCanceled);
+ }
+ }
+
+ [Test]
+ public void TestClosures()
+ {
+ TestClosure((c, t) => c.BroadcastAsync(new ComputeAction(), t));
+ TestClosure((c, t) => c.AffinityRunAsync(null, 0, new ComputeAction(), t));
+ TestClosure((c, t) => c.RunAsync(new ComputeAction(), t));
+ TestClosure((c, t) => c.RunAsync(Enumerable.Range(1, 10).Select(x => new ComputeAction()), t));
+ TestClosure((c, t) => c.CallAsync(new ComputeFunc(), t));
+ TestClosure((c, t) => c.AffinityCallAsync(null, 0, new ComputeFunc(), t));
+ TestClosure((c, t) => c.ApplyAsync(new ComputeBiFunc(), 10, t));
+ TestClosure((c, t) => c.ApplyAsync(new ComputeBiFunc(), Enumerable.Range(1, 100), t));
+ TestClosure((c, t) => c.ApplyAsync(new ComputeBiFunc(), Enumerable.Range(1, 100), new ComputeReducer(), t));
+ }
+
+ private void TestTask(Func<ICompute, CancellationToken, System.Threading.Tasks.Task> runner)
+ {
+ Job.CancelCount = 0;
+
+ TestClosure(runner);
+
+ Assert.IsTrue(TestUtils.WaitForCondition(() => Job.CancelCount > 0, 5000));
+ }
+
+ private void TestClosure(Func<ICompute, CancellationToken, System.Threading.Tasks.Task> runner)
+ {
+ using (var cts = new CancellationTokenSource())
+ {
+ var task = runner(Compute, cts.Token);
+
+ Assert.IsFalse(task.IsCanceled);
+
+ cts.Cancel();
+
+ Assert.IsTrue(task.IsCanceled);
+
+ // Pass cancelled token
+ Assert.IsTrue(runner(Compute, cts.Token).IsCanceled);
+ }
+ }
+
+ private class Task : IComputeTask<int, IList<IComputeJobResult<int>>>
+ {
+ public IDictionary<IComputeJob<int>, IClusterNode> Map(IList<IClusterNode> subgrid, object arg)
+ {
+ return Enumerable.Range(1, 100)
+ .SelectMany(x => subgrid)
+ .ToDictionary(x => (IComputeJob<int>)new Job(), x => x);
+ }
+
+ public ComputeJobResultPolicy OnResult(IComputeJobResult<int> res, IList<IComputeJobResult<int>> rcvd)
+ {
+ return ComputeJobResultPolicy.Wait;
+ }
+
+ public IList<IComputeJobResult<int>> Reduce(IList<IComputeJobResult<int>> results)
+ {
+ Assert.Fail("Reduce should not be called on a cancelled task.");
+ return results;
+ }
+ }
+
+ [Serializable]
+ private class Job : IComputeJob<int>
+ {
+ private static int _cancelCount;
+
+ public static int CancelCount
+ {
+ get { return Thread.VolatileRead(ref _cancelCount); }
+ set { Thread.VolatileWrite(ref _cancelCount, value); }
+ }
+
+ public int Execute()
+ {
+ Thread.Sleep(50);
+ return 1;
+ }
+
+ public void Cancel()
+ {
+ Interlocked.Increment(ref _cancelCount);
+ }
+ }
+
+ [Serializable]
+ private class ComputeBiFunc : IComputeFunc<int, int>
+ {
+ public int Invoke(int arg)
+ {
+ Thread.Sleep(50);
+ return arg;
+ }
+ }
+
+ private class ComputeReducer : IComputeReducer<int, int>
+ {
+ public bool Collect(int res)
+ {
+ return true;
+ }
+
+ public int Reduce()
+ {
+ return 0;
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/da601c27/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
index fe7d78f..26696b9 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
@@ -43,7 +43,7 @@ namespace Apache.Ignite.Core.Tests.Compute
private const string BinaryArgTask = "org.apache.ignite.platform.PlatformComputeBinarizableArgTask";
/** Broadcast task name. */
- private const string BroadcastTask = "org.apache.ignite.platform.PlatformComputeBroadcastTask";
+ public const string BroadcastTask = "org.apache.ignite.platform.PlatformComputeBroadcastTask";
/** Broadcast task name. */
private const string DecimalTask = "org.apache.ignite.platform.PlatformComputeDecimalTask";
[14/28] ignite git commit: ignite-1923 Test fixed
Posted by vo...@apache.org.
ignite-1923 Test fixed
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/13016460
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/13016460
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/13016460
Branch: refs/heads/ignite-2236
Commit: 130164605b2cd6142c7e01feb6aef6f3f9fac343
Parents: c160ed4
Author: agura <ag...@gridgain.com>
Authored: Mon Jan 18 17:23:47 2016 +0300
Committer: agura <ag...@gridgain.com>
Committed: Mon Jan 18 17:46:10 2016 +0300
----------------------------------------------------------------------
...IgniteCacheP2pUnmarshallingQueryErrorTest.java | 18 +++---------------
1 file changed, 3 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/13016460/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
index eafbb7a..cb7e3ad 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
@@ -24,9 +24,7 @@ import javax.cache.CacheException;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.lang.IgniteBiPredicate;
-import org.apache.ignite.marshaller.optimized.OptimizedMarshaller;
/**
* Checks behavior on exception while unmarshalling key.
@@ -80,24 +78,14 @@ public class IgniteCacheP2pUnmarshallingQueryErrorTest extends IgniteCacheP2pUnm
}
private void writeObject(ObjectOutputStream os) throws IOException {
- throw new IOException();
+ // No-op.
}
})).getAll();
- assertTrue("Request unmarshalling failed, but error response was not sent.", binaryMarshaller());
+ fail();
}
catch (Exception e) {
- assertFalse("Unexpected exception: " + e, binaryMarshaller());
+ // No-op.
}
}
-
- /**
- * @return {@code True} if binary marshaller is configured.
- */
- private boolean binaryMarshaller() {
- IgniteEx kernal = (IgniteEx)ignite(0);
-
- return !OptimizedMarshaller.class.getSimpleName().equals(kernal.context().config().getMarshaller().getClass()
- .getSimpleName());
- }
}
[07/28] ignite git commit: IGNITE-1413 .Net: Get rid of set -> map
conversion in PlatformDotNetCacheStore.writeAll. This closes #380.
Posted by vo...@apache.org.
IGNITE-1413 .Net: Get rid of set -> map conversion in PlatformDotNetCacheStore.writeAll. This closes #380.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/422272d8
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/422272d8
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/422272d8
Branch: refs/heads/ignite-2236
Commit: 422272d8298f454ac3d8aa7630799976d83fa664
Parents: 2625129
Author: Pavel Tupitsyn <pt...@gridgain.com>
Authored: Mon Jan 18 12:32:08 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 12:32:08 2016 +0300
----------------------------------------------------------------------
.../dotnet/PlatformDotNetCacheStore.java | 33 +++++---------------
.../Impl/Cache/Store/CacheStore.cs | 9 +++++-
2 files changed, 16 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/422272d8/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
index 7e65c22..47b1110 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/dotnet/PlatformDotNetCacheStore.java
@@ -224,37 +224,20 @@ public class PlatformDotNetCacheStore<K, V> implements CacheStore<K, V>, Platfor
/** {@inheritDoc} */
@SuppressWarnings({"NullableProblems", "unchecked"})
@Override public void writeAll(final Collection<Cache.Entry<? extends K, ? extends V>> entries) {
+ assert entries != null;
try {
doInvoke(new IgniteInClosureX<BinaryRawWriterEx>() {
@Override public void applyx(BinaryRawWriterEx writer) throws IgniteCheckedException {
- Map<K, V> map = new AbstractMap<K, V>() {
- @Override public int size() {
- return entries.size();
- }
-
- @Override public Set<Entry<K, V>> entrySet() {
- return new AbstractSet<Entry<K, V>>() {
- @Override public Iterator<Entry<K, V>> iterator() {
- return F.iterator(entries, new C1<Cache.Entry<? extends K, ? extends V>, Entry<K, V>>() {
- private static final long serialVersionUID = 0L;
-
- @Override public Entry<K, V> apply(Cache.Entry<? extends K, ? extends V> entry) {
- return new GridMapEntry<>(entry.getKey(), entry.getValue());
- }
- }, true);
- }
-
- @Override public int size() {
- return entries.size();
- }
- };
- }
- };
-
writer.writeByte(OP_PUT_ALL);
writer.writeLong(session());
writer.writeString(ses.cacheName());
- writer.writeMap(map);
+
+ writer.writeInt(entries.size());
+
+ for (Cache.Entry<? extends K, ? extends V> e : entries) {
+ writer.writeObject(e.getKey());
+ writer.writeObject(e.getValue());
+ }
}
}, null);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/422272d8/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs
index c66eb5e..5aa806b 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Store/CacheStore.cs
@@ -192,7 +192,14 @@ namespace Apache.Ignite.Core.Impl.Cache.Store
break;
case OpPutAll:
- _store.WriteAll(rawReader.ReadDictionary());
+ var size = rawReader.ReadInt();
+
+ var dict = new Hashtable(size);
+
+ for (int i = 0; i < size; i++)
+ dict[rawReader.ReadObject<object>()] = rawReader.ReadObject<object>();
+
+ _store.WriteAll(dict);
break;
[15/28] ignite git commit: minor
Posted by vo...@apache.org.
minor
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f6dcc63a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f6dcc63a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f6dcc63a
Branch: refs/heads/ignite-2236
Commit: f6dcc63ac10150bd99b78b482b03edcfdc767d37
Parents: da601c2
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Mon Jan 18 17:58:29 2016 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Mon Jan 18 17:58:29 2016 +0300
----------------------------------------------------------------------
.../distributed/dht/colocated/GridDhtColocatedLockFuture.java | 4 ----
1 file changed, 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f6dcc63a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
index a5f5286..cfeee4b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
@@ -1316,10 +1316,6 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
@GridToStringInclude
private Collection<KeyCacheObject> keys;
- /** Mappings to proceed. */
- @GridToStringExclude
- private Deque<GridNearLockMapping> mappings;
-
/** */
private boolean rcvRes;
[03/28] ignite git commit: IGNITE-2341: Improved warning message when
BinaryMarshaller cannot be used. Also it is not shown when
"org.apache.ignite" classes is in described situation.
Posted by vo...@apache.org.
IGNITE-2341: Improved warning message when BinaryMarshaller cannot be used. Also it is not shown when "org.apache.ignite" classes is in described situation.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/9c9b7192
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/9c9b7192
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/9c9b7192
Branch: refs/heads/ignite-2236
Commit: 9c9b7192fec102e8f086f8cbf28ba5a648b96b1d
Parents: ca0f79e
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Fri Jan 8 11:17:45 2016 +0400
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 11:36:03 2016 +0300
----------------------------------------------------------------------
.../internal/binary/BinaryClassDescriptor.java | 15 ++++++++++-----
1 file changed, 10 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/9c9b7192/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryClassDescriptor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryClassDescriptor.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryClassDescriptor.java
index 1105809..1ffa9e5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryClassDescriptor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryClassDescriptor.java
@@ -55,6 +55,9 @@ public class BinaryClassDescriptor {
/** */
public static final Unsafe UNSAFE = GridUnsafe.unsafe();
+ /** Apache Ignite base package name. */
+ private static final String OAI_PKG = "org.apache.ignite";
+
/** */
private final BinaryContext ctx;
@@ -173,11 +176,13 @@ public class BinaryClassDescriptor {
mode = serializer != null ? BinaryWriteMode.BINARY : BinaryUtils.mode(cls);
}
- if (useOptMarshaller && userType) {
- U.quietAndWarn(ctx.log(), "Class \"" + cls.getName() + "\" cannot be written in binary format because " +
- "it either implements Externalizable interface or have writeObject/readObject methods. Please " +
- "ensure that all nodes have this class in classpath. To enable binary serialization either " +
- "implement " + Binarylizable.class.getSimpleName() + " interface or set explicit serializer using " +
+ if (useOptMarshaller && userType && !cls.getName().startsWith(OAI_PKG)) {
+ U.quietAndWarn(ctx.log(), "Class \"" + cls.getName() + "\" cannot be serialized using " +
+ BinaryMarshaller.class.getSimpleName() + " because it either implements Externalizable interface " +
+ "or have writeObject/readObject methods. " + OptimizedMarshaller.class.getSimpleName() + " will be " +
+ "used instead and class instances will be deserialized on the server. Please ensure that all nodes " +
+ "have this class in classpath. To enable binary serialization either implement " +
+ Binarylizable.class.getSimpleName() + " interface or set explicit serializer using " +
"BinaryTypeConfiguration.setSerializer() method.");
}
[02/28] ignite git commit: IGNITE-2342: Set correct classlader before
calling FileSystem.get().
Posted by vo...@apache.org.
IGNITE-2342: Set correct classlader before calling FileSystem.get().
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ca0f79e6
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ca0f79e6
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ca0f79e6
Branch: refs/heads/ignite-2236
Commit: ca0f79e69ed39def0b6aa7611d9b3eaa1c7fba39
Parents: 122d27c
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Fri Jan 8 10:23:55 2016 +0400
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 11:35:53 2016 +0300
----------------------------------------------------------------------
.../hadoop/fs/BasicHadoopFileSystemFactory.java | 19 ++++++++++++++++++-
1 file changed, 18 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ca0f79e6/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/BasicHadoopFileSystemFactory.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/BasicHadoopFileSystemFactory.java b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/BasicHadoopFileSystemFactory.java
index 1e2bbf2..c791e9a 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/BasicHadoopFileSystemFactory.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/BasicHadoopFileSystemFactory.java
@@ -80,7 +80,24 @@ public class BasicHadoopFileSystemFactory implements HadoopFileSystemFactory, Ex
assert cfg != null;
try {
- return FileSystem.get(fullUri, cfg, usrName);
+ // FileSystem.get() might delegate to ServiceLoader to get the list of file system implementation.
+ // And ServiceLoader is known to be sensitive to context classloader. Therefore, we change context
+ // classloader to classloader of current class to avoid strange class-cast-exceptions.
+ ClassLoader ctxClsLdr = Thread.currentThread().getContextClassLoader();
+ ClassLoader clsLdr = getClass().getClassLoader();
+
+ if (ctxClsLdr == clsLdr)
+ return FileSystem.get(fullUri, cfg, usrName);
+ else {
+ Thread.currentThread().setContextClassLoader(clsLdr);
+
+ try {
+ return FileSystem.get(fullUri, cfg, usrName);
+ }
+ finally {
+ Thread.currentThread().setContextClassLoader(ctxClsLdr);
+ }
+ }
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
[22/28] ignite git commit: IGNITE-2384 Fixed "Notification missed in
continuous query".
Posted by vo...@apache.org.
IGNITE-2384 Fixed "Notification missed in continuous query".
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/252ba877
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/252ba877
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/252ba877
Branch: refs/heads/ignite-2236
Commit: 252ba877e2e8b357ca326ce23d8fcc83dd0138bd
Parents: 9a99633
Author: nikolay_tikhonov <nt...@gridgain.com>
Authored: Mon Jan 18 20:41:53 2016 +0300
Committer: nikolay_tikhonov <nt...@gridgain.com>
Committed: Mon Jan 18 20:42:16 2016 +0300
----------------------------------------------------------------------
.../internal/GridEventConsumeHandler.java | 3 +-
.../internal/GridMessageListenHandler.java | 6 +-
.../managers/discovery/CustomEventListener.java | 4 +-
.../discovery/GridDiscoveryManager.java | 8 +-
.../continuous/CacheContinuousQueryHandler.java | 36 +--
.../continuous/GridContinuousHandler.java | 4 +-
.../continuous/GridContinuousProcessor.java | 15 +-
...ContinuousQueryFailoverAbstractSelfTest.java | 80 ++++++
.../CacheContinuousQueryLostPartitionTest.java | 256 +++++++++++++++++++
.../IgniteBinaryCacheQueryTestSuite.java | 2 +
10 files changed, 383 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
index be35ba4..69af6cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.managers.deployment.GridDeployment;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
@@ -135,7 +136,7 @@ class GridEventConsumeHandler implements GridContinuousHandler {
}
/** {@inheritDoc} */
- @Override public void updateCounters(Map<Integer, Long> cntrs) {
+ @Override public void updateCounters(AffinityTopologyVersion topVer, Map<Integer, Long> cntrs) {
// No-op.
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java
index 2edfda5..13aeb54 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java
@@ -27,6 +27,7 @@ import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.managers.deployment.GridDeployment;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.continuous.GridContinuousBatch;
import org.apache.ignite.internal.processors.continuous.GridContinuousBatchAdapter;
import org.apache.ignite.internal.processors.continuous.GridContinuousHandler;
@@ -108,12 +109,13 @@ public class GridMessageListenHandler implements GridContinuousHandler {
}
/** {@inheritDoc} */
- @Override public void updateCounters(Map<Integer, Long> cntrs) {
+ @Override public void updateCounters(AffinityTopologyVersion topVer, Map<Integer, Long> cntrs) {
// No-op.
}
/** {@inheritDoc} */
- @Override public RegisterStatus register(UUID nodeId, UUID routineId, final GridKernalContext ctx) throws IgniteCheckedException {
+ @Override public RegisterStatus register(UUID nodeId, UUID routineId, final GridKernalContext ctx)
+ throws IgniteCheckedException {
ctx.io().addUserMessageListener(topic, pred);
return RegisterStatus.REGISTERED;
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomEventListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomEventListener.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomEventListener.java
index ab143fb..21fd842 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomEventListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomEventListener.java
@@ -18,14 +18,16 @@
package org.apache.ignite.internal.managers.discovery;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
/**
* Listener interface.
*/
public interface CustomEventListener<T extends DiscoveryCustomMessage> {
/**
+ * @param topVer Current topology version.
* @param snd Sender.
* @param msg Message.
*/
- public void onCustomEvent(ClusterNode snd, T msg);
+ public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, T msg);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 29e85dd..42f9b6d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -507,14 +507,18 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
verChanged = true;
}
+ }
+
+ nextTopVer = new AffinityTopologyVersion(topVer, minorTopVer);
+ if (type == DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT) {
for (Class cls = customMsg.getClass(); cls != null; cls = cls.getSuperclass()) {
List<CustomEventListener<DiscoveryCustomMessage>> list = customEvtLsnrs.get(cls);
if (list != null) {
for (CustomEventListener<DiscoveryCustomMessage> lsnr : list) {
try {
- lsnr.onCustomEvent(node, customMsg);
+ lsnr.onCustomEvent(nextTopVer, node, customMsg);
}
catch (Exception e) {
U.error(log, "Failed to notify direct custom event listener: " + customMsg, e);
@@ -524,8 +528,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
}
- nextTopVer = new AffinityTopologyVersion(topVer, minorTopVer);
-
// Put topology snapshot into discovery history.
// There is no race possible between history maintenance and concurrent discovery
// event notifications, since SPI notifies manager about all events from this listener.
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index ad86d65..fa54a6b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -148,6 +148,9 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/** */
private Map<Integer, Long> initUpdCntrs;
+ /** */
+ private AffinityTopologyVersion initTopVer;
+
/**
* Required by {@link Externalizable}.
*/
@@ -170,6 +173,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
* @param skipPrimaryCheck Whether to skip primary check for REPLICATED cache.
* @param taskHash Task name hash code.
* @param locCache {@code True} if local cache.
+ * @param keepBinary Keep binary flag.
*/
public CacheContinuousQueryHandler(
String cacheName,
@@ -238,7 +242,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
/** {@inheritDoc} */
- @Override public void updateCounters(Map<Integer, Long> cntrs) {
+ @Override public void updateCounters(AffinityTopologyVersion topVer, Map<Integer, Long> cntrs) {
+ this.initTopVer = topVer;
this.initUpdCntrs = cntrs;
}
@@ -382,9 +387,12 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
else {
if (!internal) {
- entry.markBackup();
+ // Skip init query and expire entries.
+ if (entry.updateCounter() != -1L) {
+ entry.markBackup();
- backupQueue.add(entry);
+ backupQueue.add(entry);
+ }
}
}
}
@@ -483,7 +491,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
e = buf.skipEntry(e);
- if (e != null)
+ if (e != null && !ctx.localNodeId().equals(nodeId))
ctx.continuous().addNotification(nodeId, routineId, e, topic, sync, true);
}
catch (ClusterTopologyCheckedException ex) {
@@ -642,15 +650,14 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
return F.asList(e);
}
- // Initial query entry or evicted entry.
- // This events should be fired immediately.
- if (e.updateCounter() == -1)
+ // Initial query entry or evicted entry. These events should be fired immediately.
+ if (e.updateCounter() == -1L)
return F.asList(e);
PartitionRecovery rec = rcvs.get(e.partition());
if (rec == null) {
- rec = new PartitionRecovery(ctx.log(getClass()), cacheContext(ctx).topology().topologyVersion(),
+ rec = new PartitionRecovery(ctx.log(getClass()), initTopVer,
initUpdCntrs == null ? null : initUpdCntrs.get(e.partition()));
PartitionRecovery oldRec = rcvs.putIfAbsent(e.partition(), rec);
@@ -724,6 +731,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
* @param initCntr Update counters.
*/
public PartitionRecovery(IgniteLogger log, AffinityTopologyVersion topVer, @Nullable Long initCntr) {
+ assert topVer.topologyVersion() > 0 : topVer;
+
this.log = log;
if (initCntr != null) {
@@ -745,17 +754,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
List<CacheContinuousQueryEntry> entries;
synchronized (pendingEvts) {
- // Received first event.
- if (curTop == AffinityTopologyVersion.NONE) {
- lastFiredEvt = entry.updateCounter();
-
- curTop = entry.topologyVersion();
-
- return F.asList(entry);
- }
-
if (curTop.compareTo(entry.topologyVersion()) < 0) {
- if (entry.updateCounter() == 1 && !entry.isBackup()) {
+ if (entry.updateCounter() == 1L && !entry.isBackup()) {
entries = new ArrayList<>(pendingEvts.size());
for (CacheContinuousQueryEntry evt : pendingEvts.values()) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java
index 900835a..8cd30a8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousHandler.java
@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.jetbrains.annotations.Nullable;
/**
@@ -154,6 +155,7 @@ public interface GridContinuousHandler extends Externalizable, Cloneable {
/**
* @param cntrs Init state for partition counters.
+ * @param topVer Topology version.
*/
- public void updateCounters(Map<Integer, Long> cntrs);
+ public void updateCounters(AffinityTopologyVersion topVer, Map<Integer, Long> cntrs);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index cb028f3..7c7e3e3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -52,6 +52,7 @@ import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.managers.discovery.CustomEventListener;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
@@ -191,7 +192,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
ctx.discovery().setCustomEventListener(StartRoutineDiscoveryMessage.class,
new CustomEventListener<StartRoutineDiscoveryMessage>() {
- @Override public void onCustomEvent(ClusterNode snd,
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer,
+ ClusterNode snd,
StartRoutineDiscoveryMessage msg) {
if (!snd.id().equals(ctx.localNodeId()) && !ctx.isStopping())
processStartRequest(snd, msg);
@@ -200,7 +202,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
ctx.discovery().setCustomEventListener(StartRoutineAckDiscoveryMessage.class,
new CustomEventListener<StartRoutineAckDiscoveryMessage>() {
- @Override public void onCustomEvent(ClusterNode snd,
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer,
+ ClusterNode snd,
StartRoutineAckDiscoveryMessage msg) {
StartFuture fut = startFuts.remove(msg.routineId());
@@ -228,7 +231,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
}
- routine.handler().updateCounters(msg.updateCounters());
+ routine.handler().updateCounters(topVer, msg.updateCounters());
}
fut.onRemoteRegistered();
@@ -246,7 +249,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
ctx.discovery().setCustomEventListener(StopRoutineDiscoveryMessage.class,
new CustomEventListener<StopRoutineDiscoveryMessage>() {
- @Override public void onCustomEvent(ClusterNode snd,
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer,
+ ClusterNode snd,
StopRoutineDiscoveryMessage msg) {
if (!snd.id().equals(ctx.localNodeId())) {
UUID routineId = msg.routineId();
@@ -265,7 +269,8 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
ctx.discovery().setCustomEventListener(StopRoutineAckDiscoveryMessage.class,
new CustomEventListener<StopRoutineAckDiscoveryMessage>() {
- @Override public void onCustomEvent(ClusterNode snd,
+ @Override public void onCustomEvent(AffinityTopologyVersion topVer,
+ ClusterNode snd,
StopRoutineAckDiscoveryMessage msg) {
StopFuture fut = stopFuts.remove(msg.routineId());
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
index 283da80..5de3d0f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
@@ -41,6 +41,9 @@ import javax.cache.CacheException;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryListenerException;
import javax.cache.event.CacheEntryUpdatedListener;
+import javax.cache.expiry.Duration;
+import javax.cache.expiry.ExpiryPolicy;
+import javax.cache.expiry.TouchedExpiryPolicy;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.MutableEntry;
import org.apache.ignite.Ignite;
@@ -90,11 +93,13 @@ import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.transactions.Transaction;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.PRIMARY;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
/**
*
@@ -1278,6 +1283,81 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
/**
* @throws Exception If failed.
*/
+ public void testBackupQueueEvict() throws Exception {
+ startGridsMultiThreaded(2);
+
+ client = true;
+
+ Ignite qryClient = startGrid(2);
+
+ CacheEventListener1 lsnr = new CacheEventListener1(false);
+
+ ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
+
+ qry.setLocalListener(lsnr);
+
+ QueryCursor<?> cur = qryClient.cache(null).query(qry);
+
+ final Collection<Object> backupQueue = backupQueue(ignite(0));
+
+ assertEquals(0, backupQueue.size());
+
+ long ttl = 100;
+
+ final ExpiryPolicy expiry = new TouchedExpiryPolicy(new Duration(MILLISECONDS, ttl));
+
+ final IgniteCache<Object, Object> cache0 = ignite(2).cache(null).withExpiryPolicy(expiry);
+
+ final List<Integer> keys = primaryKeys(ignite(1).cache(null), BACKUP_ACK_THRESHOLD);
+
+ CountDownLatch latch = new CountDownLatch(keys.size());
+
+ lsnr.latch = latch;
+
+ for (Integer key : keys) {
+ log.info("Put: " + key);
+
+ cache0.put(key, key);
+ }
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return backupQueue.isEmpty();
+ }
+ }, 2000);
+
+ assertTrue("Backup queue is not cleared: " + backupQueue, backupQueue.size() < BACKUP_ACK_THRESHOLD);
+
+ boolean wait = waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return cache0.localPeek(keys.get(0)) == null;
+ }
+ }, ttl + 1000);
+
+ assertTrue("Entry evicted.", wait);
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return backupQueue.isEmpty();
+ }
+ }, 2000);
+
+ assertTrue("Backup queue is not cleared: " + backupQueue, backupQueue.size() < BACKUP_ACK_THRESHOLD);
+
+ if (backupQueue.size() != 0) {
+ for (Object o : backupQueue) {
+ CacheContinuousQueryEntry e = (CacheContinuousQueryEntry)o;
+
+ assertNotSame("Evicted entry added to backup queue.", -1L, e.updateCounter());
+ }
+ }
+
+ cur.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testBackupQueueCleanupServerQuery() throws Exception {
Ignite qryClient = startGridsMultiThreaded(2);
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java
new file mode 100644
index 0000000..30613a4
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java
@@ -0,0 +1,256 @@
+/*
+ * 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.query.continuous;
+
+import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
+import javax.cache.event.CacheEntryCreatedListener;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryExpiredListener;
+import javax.cache.event.CacheEntryRemovedListener;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.PA;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static javax.cache.configuration.FactoryBuilder.factoryOf;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheRebalanceMode.SYNC;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC;
+
+/**
+ * Test from https://issues.apache.org/jira/browse/IGNITE-2384.
+ */
+public class CacheContinuousQueryLostPartitionTest extends GridCommonAbstractTest {
+ /** */
+ static public TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** Cache name. */
+ public static final String CACHE_NAME = "test_cache";
+
+ /** Cache name. */
+ public static final String TX_CACHE_NAME = "tx_test_cache";
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ startGridsMultiThreaded(2);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxEvent() throws Exception {
+ testEvent(TX_CACHE_NAME, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicEvent() throws Exception {
+ testEvent(CACHE_NAME, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxClientEvent() throws Exception {
+ testEvent(TX_CACHE_NAME, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicClientEvent() throws Exception {
+ testEvent(CACHE_NAME, true);
+ }
+
+ /**
+ * @param cacheName Cache name.
+ * @throws Exception If failed.
+ */
+ public void testEvent(String cacheName, boolean client) throws Exception {
+ IgniteCache<Integer, String> cache1 = grid(0).getOrCreateCache(cacheName);
+
+ final AllEventListener<Integer, String> lsnr1 = registerCacheListener(cache1);
+
+ IgniteCache<Integer, String> cache2 = grid(1).getOrCreateCache(cacheName);
+
+ Integer key = primaryKey(cache1);
+
+ cache1.put(key, "1");
+
+ // Note the issue is only reproducible if the second registration is done right
+ // here, after the first put() above.
+ AllEventListener<Integer, String> lsnr20;
+
+ if (client) {
+ IgniteCache<Integer, String> clnCache = startGrid(3).getOrCreateCache(cacheName);
+
+ lsnr20 = registerCacheListener(clnCache);
+ }
+ else
+ lsnr20 = registerCacheListener(cache2);
+
+ final AllEventListener<Integer, String> lsnr2 = lsnr20;
+
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return lsnr1.createdCnt.get() == 1;
+ }
+ }, 2000L) : "Unexpected number of events: " + lsnr1.createdCnt.get();
+
+ // Sanity check.
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return lsnr2.createdCnt.get() == 0;
+ }
+ }, 2000L) : "Expected no create events, but got: " + lsnr2.createdCnt.get();
+
+ // node2 now becomes the primary for the key.
+ grid(0).close();
+
+ cache2.put(key, "2");
+
+ // Sanity check.
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return lsnr1.createdCnt.get() == 1;
+ }
+ }, 2000L) : "Expected no change here, but got: " + lsnr1.createdCnt.get();
+
+ // Sanity check.
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return lsnr2.updatedCnt.get() == 0;
+ }
+ }, 2000L) : "Expected no update events, but got: " + lsnr2.updatedCnt.get();
+
+ System.out.println(">>>>> " + lsnr2.createdCnt.get());
+
+ // This assertion fails: 0 events get delivered.
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return lsnr2.createdCnt.get() == 1;
+ }
+ }, 2000L) : "Expected a single event due to '2', but got: " + lsnr2.createdCnt.get();
+ }
+
+ /**
+ * @param cache Cache.
+ * @return Event listener.
+ */
+ private AllEventListener<Integer, String> registerCacheListener(IgniteCache<Integer, String> cache) {
+ AllEventListener<Integer, String> lsnr = new AllEventListener<>();
+
+ cache.registerCacheEntryListener(
+ new MutableCacheEntryListenerConfiguration<>(factoryOf(lsnr), null, true, false));
+
+ return lsnr;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String name) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(name);
+
+ TcpDiscoverySpi spi = new TcpDiscoverySpi();
+
+ spi.setIpFinder(ipFinder);
+
+ cfg.setDiscoverySpi(spi);
+ cfg.setCacheConfiguration(cache(TX_CACHE_NAME), cache(CACHE_NAME));
+
+ if (name.endsWith("3"))
+ cfg.setClientMode(true);
+
+ return cfg;
+ }
+
+ /**
+ * @param cacheName Cache name.
+ * @return Cache configuration.
+ */
+ protected CacheConfiguration<Integer, String> cache(String cacheName) {
+ CacheConfiguration<Integer, String> cfg = new CacheConfiguration<>(cacheName);
+
+ cfg.setCacheMode(PARTITIONED);
+
+ if (cacheName.equals(CACHE_NAME))
+ cfg.setAtomicityMode(ATOMIC);
+ else
+ cfg.setAtomicityMode(TRANSACTIONAL);
+
+ cfg.setRebalanceMode(SYNC);
+ cfg.setWriteSynchronizationMode(PRIMARY_SYNC);
+ cfg.setBackups(0);
+
+ return cfg;
+ }
+
+ /**
+ * Event listener.
+ */
+ public static class AllEventListener<K, V> implements CacheEntryCreatedListener<K, V>,
+ CacheEntryUpdatedListener<K, V>, CacheEntryRemovedListener<K, V>, CacheEntryExpiredListener<K, V>,
+ Serializable {
+ /** */
+ final AtomicInteger createdCnt = new AtomicInteger();
+
+ /** */
+ final AtomicInteger updatedCnt = new AtomicInteger();
+
+ /** {@inheritDoc} */
+ @Override public void onCreated(Iterable<CacheEntryEvent<? extends K, ? extends V>> evts) {
+ createdCnt.incrementAndGet();
+
+ System.out.printf("onCreate: %s. \n", evts);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onExpired(Iterable<CacheEntryEvent<? extends K, ? extends V>> evts) {
+ System.out.printf("onExpired: %s. \n", evts);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onRemoved(Iterable<CacheEntryEvent<? extends K, ? extends V>> evts) {
+ System.out.printf("onRemoved: %s. \n", evts);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends K, ? extends V>> evts) {
+ updatedCnt.incrementAndGet();
+
+ System.out.printf("onUpdated: %s.", evts);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/252ba877/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
index eddfcf4..3bab1f9 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
@@ -75,6 +75,7 @@ import org.apache.ignite.internal.processors.cache.distributed.replicated.Ignite
import org.apache.ignite.internal.processors.cache.local.IgniteCacheLocalAtomicQuerySelfTest;
import org.apache.ignite.internal.processors.cache.local.IgniteCacheLocalFieldsQuerySelfTest;
import org.apache.ignite.internal.processors.cache.local.IgniteCacheLocalQuerySelfTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryLostPartitionTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFailoverAtomicPrimaryWriteOrderSelfTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFailoverAtomicReplicatedSelfTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFailoverTxReplicatedSelfTest;
@@ -200,6 +201,7 @@ public class IgniteBinaryCacheQueryTestSuite extends TestSuite {
suite.addTestSuite(CacheContinuousQueryFailoverAtomicReplicatedSelfTest.class);
suite.addTestSuite(CacheContinuousQueryFailoverTxSelfTest.class);
suite.addTestSuite(CacheContinuousQueryFailoverTxReplicatedSelfTest.class);
+ suite.addTestSuite(CacheContinuousQueryLostPartitionTest.class);
// Reduce fields queries.
suite.addTestSuite(GridCacheReduceFieldsQueryLocalSelfTest.class);
[06/28] ignite git commit: IGNITE-2396 - Added services failing test.
Posted by vo...@apache.org.
IGNITE-2396 - Added services failing test.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/9d139e53
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/9d139e53
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/9d139e53
Branch: refs/heads/ignite-2236
Commit: 9d139e53165feb6b1641a1b4da37fcb8721050e2
Parents: 2625129
Author: Alexey Goncharuk <al...@gmail.com>
Authored: Mon Jan 18 12:26:18 2016 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Mon Jan 18 12:26:18 2016 +0300
----------------------------------------------------------------------
.../IgniteServiceDynamicCachesSelfTest.java | 183 +++++++++++++++++++
1 file changed, 183 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/9d139e53/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java
new file mode 100644
index 0000000..c41f2f0
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/IgniteServiceDynamicCachesSelfTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.service;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.IgniteServices;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.PA;
+import org.apache.ignite.resources.LoggerResource;
+import org.apache.ignite.services.Service;
+import org.apache.ignite.services.ServiceContext;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class IgniteServiceDynamicCachesSelfTest extends GridCommonAbstractTest {
+ /** */
+ private static final int GRID_CNT = 4;
+
+ /** */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+
+ discoSpi.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ startGrids(GRID_CNT);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDeployCalledAfterCacheStart() throws Exception {
+ String cacheName = "cache";
+
+ CacheConfiguration ccfg = new CacheConfiguration(cacheName);
+ ccfg.setBackups(1);
+
+ Ignite ig = ignite(0);
+
+ ig.createCache(ccfg);
+
+ try {
+ final IgniteServices svcs = ig.services();
+
+ final String svcName = "myService";
+
+ svcs.deployKeyAffinitySingleton(svcName, new TestService(), cacheName, "key");
+
+ boolean res = GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return svcs.service(svcName) != null;
+ }
+ }, 10 * 1000);
+
+ assertTrue("Service was not deployed", res);
+
+ ig.destroyCache(cacheName);
+
+ res = GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return svcs.service(svcName) == null;
+ }
+ }, 10 * 1000);
+
+ assertTrue("Service was not undeployed", res);
+ }
+ finally {
+ ig.services().cancelAll();
+
+ ig.destroyCache(cacheName);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDeployCalledBeforeCacheStart() throws Exception {
+ String cacheName = "cache";
+
+ CacheConfiguration ccfg = new CacheConfiguration(cacheName);
+ ccfg.setBackups(1);
+
+ Ignite ig = ignite(0);
+
+ final IgniteServices svcs = ig.services();
+
+ final String svcName = "myService";
+
+ svcs.deployKeyAffinitySingleton(svcName, new TestService(), cacheName, "key");
+
+ assert svcs.service(svcName) == null;
+
+ ig.createCache(ccfg);
+
+ try {
+ boolean res = GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return svcs.service(svcName) != null;
+ }
+ }, 10 * 1000);
+
+ assertTrue("Service was not deployed", res);
+
+ ig.destroyCache(cacheName);
+
+ res = GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return svcs.service(svcName) == null;
+ }
+ }, 10 * 1000);
+
+ assertTrue("Service was not undeployed", res);
+ }
+ finally {
+ ig.services().cancelAll();
+
+ ig.destroyCache(cacheName);
+ }
+ }
+
+ /**
+ *
+ */
+ private static class TestService implements Service {
+ /** */
+ @LoggerResource
+ private IgniteLogger log;
+
+ /** {@inheritDoc} */
+ @Override public void cancel(ServiceContext ctx) {
+ log.info("Service cancelled.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void init(ServiceContext ctx) throws Exception {
+ log.info("Service deployed.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void execute(ServiceContext ctx) throws Exception {
+ log.info("Service executed.");
+ }
+ }
+}
[23/28] ignite git commit: IGNITE-1993 Fixed "JDBC discovery uses
non-standard SQL when creating table (not compatible with Oracle)"
Posted by vo...@apache.org.
IGNITE-1993 Fixed "JDBC discovery uses non-standard SQL when creating table (not compatible with Oracle)"
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/7ca1a8db
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/7ca1a8db
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/7ca1a8db
Branch: refs/heads/ignite-2236
Commit: 7ca1a8db636fafe9908f5e413d1066a22f7c3d54
Parents: 252ba87
Author: Nigel Westbury <ni...@1spatial.com>
Authored: Mon Jan 18 21:29:19 2016 +0300
Committer: nikolay_tikhonov <nt...@gridgain.com>
Committed: Mon Jan 18 21:29:19 2016 +0300
----------------------------------------------------------------------
.../ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java | 48 ++++++++++++++++----
1 file changed, 38 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/7ca1a8db/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java
index 69fa3f8..9d25931 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java
@@ -19,6 +19,7 @@ package org.apache.ignite.spi.discovery.tcp.ipfinder.jdbc;
import java.net.InetSocketAddress;
import java.sql.Connection;
+import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -58,6 +59,10 @@ import static java.sql.Connection.TRANSACTION_READ_COMMITTED;
* The database will contain 1 table which will hold IP addresses.
*/
public class TcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
+ /** Name of the address table, in upper case. Mostly table names are not case-sensitive
+ * but databases such as Oracle require table names in upper-case when looking them up in the metadata. */
+ public static final String ADDRS_TABLE_NAME = "TBL_ADDRS";
+
/** Query to get addresses. */
public static final String GET_ADDRS_QRY = "select hostname, port from tbl_addrs";
@@ -69,7 +74,7 @@ public class TcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
/** Query to create addresses table. */
public static final String CREATE_ADDRS_TABLE_QRY =
- "create table if not exists tbl_addrs (" +
+ "create table tbl_addrs (" +
"hostname VARCHAR(1024), " +
"port INT)";
@@ -290,8 +295,6 @@ public class TcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
Connection conn = null;
- Statement stmt = null;
-
boolean committed = false;
try {
@@ -301,12 +304,38 @@ public class TcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
- // Create tbl_addrs.
- stmt = conn.createStatement();
-
- stmt.executeUpdate(CREATE_ADDRS_TABLE_QRY);
-
- conn.commit();
+ DatabaseMetaData dbm = conn.getMetaData();
+
+ // Many JDBC implementations support an 'if not exists' clause
+ // in the create statement which will check and create atomically.
+ // However not all databases support it, for example Oracle,
+ // so we do not use it.
+ try (ResultSet tables = dbm.getTables(null, null, ADDRS_TABLE_NAME, null)) {
+ if (!tables.next()) {
+ // Table does not exist
+ // Create tbl_addrs.
+ try (Statement stmt = conn.createStatement()) {
+ stmt.executeUpdate(CREATE_ADDRS_TABLE_QRY);
+
+ conn.commit();
+ }
+ catch (SQLException e) {
+ // Due to a race condition, the table may have been
+ // created since we tested above for its existence.
+ // We must ignore the exception if this is the
+ // cause.
+ // However different JDBC driver implementations may
+ // return different codes and messages in the
+ // exception, so the safest way to determine if this
+ // exception is to be ignored is to test again to
+ // see if the table has been created.
+ try (ResultSet tablesAgain = dbm.getTables(null, null, ADDRS_TABLE_NAME, null)) {
+ if (!tablesAgain.next())
+ throw e;
+ }
+ }
+ }
+ }
committed = true;
@@ -322,7 +351,6 @@ public class TcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
if (!committed)
U.rollbackConnectionQuiet(conn);
- U.closeQuiet(stmt);
U.closeQuiet(conn);
initLatch.countDown();
[11/28] ignite git commit: Disabled failing tests.
Posted by vo...@apache.org.
Disabled failing tests.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/36486b41
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/36486b41
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/36486b41
Branch: refs/heads/ignite-2236
Commit: 36486b4117a7447802dbcb73caddad423d980fde
Parents: da601c2
Author: sboikov <sb...@gridgain.com>
Authored: Mon Jan 18 16:20:57 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Mon Jan 18 16:20:57 2016 +0300
----------------------------------------------------------------------
.../tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java | 2 ++
.../cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java | 2 ++
2 files changed, 4 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/36486b41/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
index b7c0deb..25e3611 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationRecoveryAckClosureSelfTest.java
@@ -244,6 +244,8 @@ public class IgniteTcpCommunicationRecoveryAckClosureSelfTest<T extends Communic
* @throws Exception If failed.
*/
public void testQueueOverflow() throws Exception {
+ fail("https://issues.apache.org/jira/browse/IGNITE-172");
+
for (int i = 0; i < 3; i++) {
try {
startSpis(5, 60_000, 10);
http://git-wip-us.apache.org/repos/asf/ignite/blob/36486b41/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
index 8aaf34b..eafbb7a 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheP2pUnmarshallingQueryErrorTest.java
@@ -65,6 +65,8 @@ public class IgniteCacheP2pUnmarshallingQueryErrorTest extends IgniteCacheP2pUnm
* @throws Exception If failed.
*/
public void testResponseMessageOnRequestUnmarshallingFailed() throws Exception {
+ fail("https://issues.apache.org/jira/browse/IGNITE-1923");
+
readCnt.set(Integer.MAX_VALUE);
try {
[19/28] ignite git commit: IGNITE-2166: .NET: Added toBuilder()
method to BinaryObject interface. This closes #379.
Posted by vo...@apache.org.
IGNITE-2166: .NET: Added toBuilder() method to BinaryObject interface. This closes #379.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/22cf3a34
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/22cf3a34
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/22cf3a34
Branch: refs/heads/ignite-2236
Commit: 22cf3a345e8f6bd517d4b0e875cabdd69491713e
Parents: c160ed4
Author: Pavel Tupitsyn <pt...@gridgain.com>
Authored: Mon Jan 18 18:31:39 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 18:31:39 2016 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/binary/BinaryEnumObjectImpl.java | 3 +--
.../Apache.Ignite.Core.Tests/Binary/BinaryBuilderSelfTest.cs | 8 ++++----
.../dotnet/Apache.Ignite.Core/Binary/IBinaryObject.cs | 8 ++++++++
.../dotnet/Apache.Ignite.Core/Impl/Binary/Binary.cs | 3 +++
.../dotnet/Apache.Ignite.Core/Impl/Binary/BinaryEnum.cs | 6 ++++++
.../dotnet/Apache.Ignite.Core/Impl/Binary/BinaryObject.cs | 6 ++++++
6 files changed, 28 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/22cf3a34/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java
index 536c582..180e20a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java
@@ -23,7 +23,6 @@ import org.apache.ignite.binary.BinaryObjectBuilder;
import org.apache.ignite.binary.BinaryObjectException;
import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.internal.GridDirectTransient;
-import org.apache.ignite.internal.binary.builder.BinaryObjectBuilderImpl;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheObjectContext;
import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
@@ -124,7 +123,7 @@ public class BinaryEnumObjectImpl implements BinaryObjectEx, Externalizable, Cac
/** {@inheritDoc} */
@Override public BinaryObjectBuilder toBuilder() throws BinaryObjectException {
- return BinaryObjectBuilderImpl.wrap(this);
+ throw new UnsupportedOperationException("Builder cannot be created for enum.");
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/22cf3a34/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/BinaryBuilderSelfTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/BinaryBuilderSelfTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/BinaryBuilderSelfTest.cs
index 373e173..d442fb1 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/BinaryBuilderSelfTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Binary/BinaryBuilderSelfTest.cs
@@ -265,7 +265,7 @@ namespace Apache.Ignite.Core.Tests.Binary
Assert.AreEqual(0, meta.Fields.Count);
// Populate it with field.
- IBinaryObjectBuilder builder = _grid.GetBinary().GetBuilder(binObj);
+ IBinaryObjectBuilder builder = binObj.ToBuilder();
Assert.IsNull(builder.GetField<object>("val"));
@@ -288,7 +288,7 @@ namespace Apache.Ignite.Core.Tests.Binary
Assert.AreEqual(BinaryTypeNames.TypeNameObject, meta.GetFieldTypeName("val"));
// Perform field remove.
- builder = _grid.GetBinary().GetBuilder(binObj);
+ builder = binObj.ToBuilder();
Assert.AreEqual(val, builder.GetField<object>("val"));
@@ -314,7 +314,7 @@ namespace Apache.Ignite.Core.Tests.Binary
.SetField("val2", inner)
.Build();
- binObj = _grid.GetBinary().GetBuilder(binObj).RemoveField("val").Build();
+ binObj = binObj.ToBuilder().RemoveField("val").Build();
Remove obj = binObj.Deserialize<Remove>();
@@ -636,7 +636,7 @@ namespace Apache.Ignite.Core.Tests.Binary
Assert.AreEqual(6, obj.FDouble);
// Overwrite.
- binObj = _grid.GetBinary().GetBuilder(binObj)
+ binObj = binObj.ToBuilder()
.SetField<byte>("fByte", 7)
.SetField("fBool", false)
.SetField<short>("fShort", 8)
http://git-wip-us.apache.org/repos/asf/ignite/blob/22cf3a34/modules/platforms/dotnet/Apache.Ignite.Core/Binary/IBinaryObject.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Binary/IBinaryObject.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Binary/IBinaryObject.cs
index c5aa80e..841972d 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Binary/IBinaryObject.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Binary/IBinaryObject.cs
@@ -63,5 +63,13 @@ namespace Apache.Ignite.Core.Binary
/// The value of underlying enum in int form.
/// </value>
int EnumValue { get; }
+
+ /// <summary>
+ /// Creates a new <see cref="IBinaryObjectBuilder"/> based on this object.
+ /// <para />
+ /// This is equivalent to <see cref="IBinary.GetBuilder(IBinaryObject)"/>.
+ /// </summary>
+ /// <returns>New <see cref="IBinaryObjectBuilder"/> based on this object.</returns>
+ IBinaryObjectBuilder ToBuilder();
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/22cf3a34/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Binary.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Binary.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Binary.cs
index 43a4bb8..7062606 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Binary.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/Binary.cs
@@ -104,6 +104,9 @@ namespace Apache.Ignite.Core.Impl.Binary
if (obj0 == null)
throw new ArgumentException("Unsupported object type: " + obj.GetType());
+ if (obj0 is BinaryEnum)
+ throw new InvalidOperationException("Builder cannot be created for enum.");
+
IBinaryTypeDescriptor desc = _marsh.GetDescriptor(true, obj0.TypeId);
return Builder0(null, obj0, desc);
http://git-wip-us.apache.org/repos/asf/ignite/blob/22cf3a34/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryEnum.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryEnum.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryEnum.cs
index 97f44b0..50b2eb8 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryEnum.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryEnum.cs
@@ -87,6 +87,12 @@ namespace Apache.Ignite.Core.Impl.Binary
}
/** <inheritdoc /> */
+ public IBinaryObjectBuilder ToBuilder()
+ {
+ return _marsh.Ignite.GetBinary().GetBuilder(this);
+ }
+
+ /** <inheritdoc /> */
public bool Equals(BinaryEnum other)
{
if (ReferenceEquals(null, other))
http://git-wip-us.apache.org/repos/asf/ignite/blob/22cf3a34/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryObject.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryObject.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryObject.cs
index 90607dd..16f95a1 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryObject.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Binary/BinaryObject.cs
@@ -133,6 +133,12 @@ namespace Apache.Ignite.Core.Impl.Binary
}
}
+ /** <inheritdoc /> */
+ public IBinaryObjectBuilder ToBuilder()
+ {
+ return _marsh.Ignite.GetBinary().GetBuilder(this);
+ }
+
/// <summary>
/// Internal deserialization routine.
/// </summary>
[12/28] ignite git commit: .NET: IgniteFutureCancelledException is
marked as serializable.
Posted by vo...@apache.org.
.NET: IgniteFutureCancelledException is marked as serializable.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ce2dda7c
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ce2dda7c
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ce2dda7c
Branch: refs/heads/ignite-2236
Commit: ce2dda7ca9a9be34458a4b472d9c4e34eba1a94e
Parents: 36486b4
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Mon Jan 18 17:24:18 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 17:24:18 2016 +0300
----------------------------------------------------------------------
.../Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ce2dda7c/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs
index 02433ce..cc9436d 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs
@@ -23,6 +23,7 @@ namespace Apache.Ignite.Core.Common
/// <summary>
/// Indicates future cancellation within Ignite.
/// </summary>
+ [Serializable]
public class IgniteFutureCancelledException : IgniteException
{
/// <summary>
[26/28] ignite git commit: IGNITE-2236: Minor.
Posted by vo...@apache.org.
IGNITE-2236: Minor.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/046690a1
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/046690a1
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/046690a1
Branch: refs/heads/ignite-2236
Commit: 046690a111466a046280f471b342a4218968c272
Parents: fd574b6
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Jan 19 11:31:32 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Jan 19 11:31:32 2016 +0300
----------------------------------------------------------------------
.../processors/cache/distributed/near/GridNearTxFinishFuture.java | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/046690a1/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index 4581121..986fa9a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@ -28,7 +28,6 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
[16/28] ignite git commit: Merge remote-tracking branch
'origin/master'
Posted by vo...@apache.org.
Merge remote-tracking branch 'origin/master'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d85616b9
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d85616b9
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d85616b9
Branch: refs/heads/ignite-2236
Commit: d85616b9bd788d0192272a13df18f9319f14a836
Parents: f6dcc63 1301646
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Mon Jan 18 17:58:48 2016 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Mon Jan 18 17:58:48 2016 +0300
----------------------------------------------------------------------
.../internal/processors/igfs/IgfsProcessor.java | 3 ---
...CommunicationRecoveryAckClosureSelfTest.java | 2 ++
...niteCacheP2pUnmarshallingQueryErrorTest.java | 20 +++++---------------
.../Common/IgniteFutureCancelledException.cs | 1 +
4 files changed, 8 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
[13/28] ignite git commit: IGNITE-1788: Removed duplicate check of a
single invariant from IgfsProcessor. This closes #221.
Posted by vo...@apache.org.
IGNITE-1788: Removed duplicate check of a single invariant from IgfsProcessor. This closes #221.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c160ed47
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c160ed47
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c160ed47
Branch: refs/heads/ignite-2236
Commit: c160ed4798c9d86e870a39813f7ec2037f8da601
Parents: ce2dda7
Author: iveselovskiy <iv...@gridgain.com>
Authored: Mon Jan 18 17:28:42 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 17:28:42 2016 +0300
----------------------------------------------------------------------
.../org/apache/ignite/internal/processors/igfs/IgfsProcessor.java | 3 ---
1 file changed, 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c160ed47/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsProcessor.java
index 5b8cf86..21446e1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsProcessor.java
@@ -297,9 +297,6 @@ public class IgfsProcessor extends IgfsProcessorAdapter {
if (GridQueryProcessor.isEnabled(metaCacheCfg))
throw new IgniteCheckedException("IGFS metadata cache cannot start with enabled query indexing.");
- if (GridQueryProcessor.isEnabled(metaCacheCfg))
- throw new IgniteCheckedException("IGFS metadata cache cannot start with enabled query indexing.");
-
if (metaCacheCfg.getAtomicityMode() != TRANSACTIONAL)
throw new IgniteCheckedException("Meta cache should be transactional: " + cfg.getMetaCacheName());
[04/28] ignite git commit: IGNITE-2340: Improved error thrown when
PROXY mode exists,
but secondary file system is not IgniteHadoopIgfsSecondaryFileSystem.
Posted by vo...@apache.org.
IGNITE-2340: Improved error thrown when PROXY mode exists, but secondary file system is not IgniteHadoopIgfsSecondaryFileSystem.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b20dc40f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b20dc40f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b20dc40f
Branch: refs/heads/ignite-2236
Commit: b20dc40feba82424934e11afa7d2baded1fb06bb
Parents: 9c9b719
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Fri Jan 8 11:26:03 2016 +0400
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Jan 18 11:36:17 2016 +0300
----------------------------------------------------------------------
.../apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java | 7 ++++++-
.../apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java | 7 +++++++
2 files changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/b20dc40f/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
index 71f6435..45b968c 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
@@ -32,7 +32,9 @@ import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.Progressable;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.configuration.FileSystemConfiguration;
import org.apache.ignite.hadoop.fs.HadoopFileSystemFactory;
+import org.apache.ignite.hadoop.fs.IgniteHadoopIgfsSecondaryFileSystem;
import org.apache.ignite.igfs.IgfsBlockLocation;
import org.apache.ignite.igfs.IgfsException;
import org.apache.ignite.igfs.IgfsFile;
@@ -335,7 +337,10 @@ public class IgniteHadoopFileSystem extends FileSystem {
throw new IOException("Failed to get secondary file system factory.", e);
}
- assert factory != null;
+ if (factory == null)
+ throw new IOException("Failed to get secondary file system factory (did you set " +
+ IgniteHadoopIgfsSecondaryFileSystem.class.getName() + " as \"secondaryFIleSystem\" in " +
+ FileSystemConfiguration.class.getName() + "?)");
if (factory instanceof LifecycleAware)
((LifecycleAware) factory).start();
http://git-wip-us.apache.org/repos/asf/ignite/blob/b20dc40f/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
index 0d7de86..ac457a4 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
@@ -38,7 +38,9 @@ import org.apache.hadoop.util.DataChecksum;
import org.apache.hadoop.util.Progressable;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.configuration.FileSystemConfiguration;
import org.apache.ignite.hadoop.fs.HadoopFileSystemFactory;
+import org.apache.ignite.hadoop.fs.IgniteHadoopIgfsSecondaryFileSystem;
import org.apache.ignite.igfs.IgfsBlockLocation;
import org.apache.ignite.igfs.IgfsFile;
import org.apache.ignite.igfs.IgfsMode;
@@ -344,6 +346,11 @@ public class IgniteHadoopFileSystem extends AbstractFileSystem implements Closea
throw new IOException("Failed to get secondary file system factory.", e);
}
+ if (factory == null)
+ throw new IOException("Failed to get secondary file system factory (did you set " +
+ IgniteHadoopIgfsSecondaryFileSystem.class.getName() + " as \"secondaryFIleSystem\" in " +
+ FileSystemConfiguration.class.getName() + "?)");
+
assert factory != null;
if (factory instanceof LifecycleAware)
[28/28] ignite git commit: IGNITE-2236: Removed TODOs.
Posted by vo...@apache.org.
IGNITE-2236: Removed TODOs.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/991eef2a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/991eef2a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/991eef2a
Branch: refs/heads/ignite-2236
Commit: 991eef2a3cf3e05347d9310ebc04507c684c6dc6
Parents: 68e3f44
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Jan 19 12:12:00 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Jan 19 12:12:00 2016 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/util/future/GridCompoundFuture.java | 3 ---
1 file changed, 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/991eef2a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
index c7812b7..c382497 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
@@ -54,17 +54,14 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> implements Ig
AtomicIntegerFieldUpdater.newUpdater(GridCompoundFuture.class, "lsnrCalls");
/** Futures. */
- // TODO: Size?
protected final ArrayList<IgniteInternalFuture<T>> futs = new ArrayList<>();
/** Reducer. */
@GridToStringInclude
- // TODO: To extension.
private final IgniteReducer<T, R> rdc;
/** Initialization flag. Updated via {@link #FLAGS_UPD}. */
@SuppressWarnings("unused")
- // TODO: Merge to "lsnrCalls"
private volatile int initFlag;
/** Listener calls. Updated via {@link #LSNR_CALLS_UPD}. */
[27/28] ignite git commit: IGNITE-2236: TODOs for potential optos.
Posted by vo...@apache.org.
IGNITE-2236: TODOs for potential optos.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/68e3f444
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/68e3f444
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/68e3f444
Branch: refs/heads/ignite-2236
Commit: 68e3f444b0dc84ba24736b96bc7eab4d126a09fd
Parents: 046690a
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Jan 19 11:42:04 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Jan 19 11:42:04 2016 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/util/future/GridCompoundFuture.java | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/68e3f444/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
index c382497..c7812b7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
@@ -54,14 +54,17 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> implements Ig
AtomicIntegerFieldUpdater.newUpdater(GridCompoundFuture.class, "lsnrCalls");
/** Futures. */
+ // TODO: Size?
protected final ArrayList<IgniteInternalFuture<T>> futs = new ArrayList<>();
/** Reducer. */
@GridToStringInclude
+ // TODO: To extension.
private final IgniteReducer<T, R> rdc;
/** Initialization flag. Updated via {@link #FLAGS_UPD}. */
@SuppressWarnings("unused")
+ // TODO: Merge to "lsnrCalls"
private volatile int initFlag;
/** Listener calls. Updated via {@link #LSNR_CALLS_UPD}. */