This is an automated email from the ASF dual-hosted git repository.
alexpl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 1873228 IGNITE-13106 Java thin client: Fix race between response and
notification for compute tasks - Fixes #7885.
1873228 is described below
commit 1873228a8b63b2347e2dedbd2c042ca25d7bb7ad
Author: Aleksey Plekhanov <plehanov.a...@gmail.com>
AuthorDate: Mon Jun 15 11:25:44 2020 +0500
IGNITE-13106 Java thin client: Fix race between response and notification
for compute tasks - Fixes #7885.
Signed-off-by: Aleksey Plekhanov <plehanov.a...@gmail.com>
---
.../internal/client/thin/ClientComputeImpl.java | 24 +++++++++++++++++++---
.../internal/client/thin/ComputeTaskTest.java | 20 ++++++++++--------
2 files changed, 33 insertions(+), 11 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientComputeImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientComputeImpl.java
index 6238e52..3d53c010 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientComputeImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientComputeImpl.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.binary.BinaryWriterExImpl;
import org.apache.ignite.internal.binary.streams.BinaryHeapInputStream;
+import org.apache.ignite.internal.processors.platform.client.ClientStatus;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
@@ -207,7 +208,12 @@ class ClientComputeImpl implements ClientCompute,
NotificationListener {
if (task == null) // Channel is closed concurrently, retry with
another channel.
continue;
- task.fut.listen(f -> removeTask(task.ch, task.taskId));
+ task.fut.listen(f -> {
+ // Don't remove task if future was canceled by user. This task
can be added again later by notification.
+ // To prevent leakage tasks for cancelled futures will be
removed on notification (or channel close event).
+ if (!f.isCancelled())
+ removeTask(task.ch, task.taskId);
+ });
return new ClientFutureImpl<>((GridFutureAdapter<R>)task.fut);
}
@@ -257,13 +263,16 @@ class ClientComputeImpl implements ClientCompute,
NotificationListener {
if (op == ClientOperation.COMPUTE_TASK_FINISHED) {
Object res = payload == null ? null : utils.readObject(new
BinaryHeapInputStream(payload), false);
- ClientComputeTask<Object> task = removeTask(ch, rsrcId);
+ ClientComputeTask<Object> task = addTask(ch, rsrcId);
if (task != null) { // If channel is closed concurrently, task is
already done with "channel closed" reason.
if (err == null)
task.fut.onDone(res);
else
task.fut.onDone(err);
+
+ if (task.fut.isCancelled())
+ removeTask(ch, rsrcId);
}
}
}
@@ -410,7 +419,16 @@ class ClientComputeImpl implements ClientCompute,
NotificationListener {
fut = new GridFutureAdapter<R>() {
@Override public boolean cancel() {
if (onCancelled()) {
- ch.service(RESOURCE_CLOSE, req ->
req.out().writeLong(taskId), null);
+ try {
+ ch.service(RESOURCE_CLOSE, req ->
req.out().writeLong(taskId), null);
+ }
+ catch (ClientServerError e) {
+ // Ignore "resource doesn't exist" error. The task
can be completed concurrently on the
+ // server, but we already complete future with
"cancelled" state, so result will never be
+ // received by a client.
+ if (e.getCode() !=
ClientStatus.RESOURCE_DOES_NOT_EXIST)
+ throw new ClientException(e);
+ }
return true;
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ComputeTaskTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ComputeTaskTest.java
index d400daf..4e13f49 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ComputeTaskTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ComputeTaskTest.java
@@ -33,7 +33,6 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteException;
import org.apache.ignite.Ignition;
@@ -56,7 +55,6 @@ import org.apache.ignite.mxbean.ClientProcessorMXBean;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jetbrains.annotations.Nullable;
-import org.junit.Ignore;
import org.junit.Test;
/**
@@ -192,7 +190,8 @@ public class ComputeTaskTest extends GridCommonAbstractTest
{
fut.cancel(true);
-
assertTrue(((ClientComputeImpl)client.compute()).activeTaskFutures().isEmpty());
+ assertTrue(GridTestUtils.waitForCondition(
+ () ->
((ClientComputeImpl)client.compute()).activeTaskFutures().isEmpty(), TIMEOUT));
assertTrue(fut.isCancelled());
assertTrue(fut.isDone());
@@ -343,7 +342,7 @@ public class ComputeTaskTest extends GridCommonAbstractTest
{
compute.execute(TestTask.class.getName(), null);
- assertEquals(1, compute.activeTaskFutures().size());
+ assertTrue(GridTestUtils.waitForCondition(() ->
compute.activeTaskFutures().size() == 1, TIMEOUT));
assertTrue(fut1.isDone());
@@ -359,7 +358,7 @@ public class ComputeTaskTest extends GridCommonAbstractTest
{
fut3.get(TIMEOUT, TimeUnit.MILLISECONDS);
- assertTrue(compute.activeTaskFutures().isEmpty());
+ assertTrue(GridTestUtils.waitForCondition(() ->
compute.activeTaskFutures().isEmpty(), TIMEOUT));
}
}
@@ -481,7 +480,6 @@ public class ComputeTaskTest extends GridCommonAbstractTest
{
*
*/
@Test
- @Ignore("https://issues.apache.org/jira/browse/IGNITE-12845";)
public void testExecuteTaskConcurrentLoad() throws Exception {
try (IgniteClient client = startClient(0)) {
int threadsCnt = 20;
@@ -511,9 +509,14 @@ public class ComputeTaskTest extends
GridCommonAbstractTest {
Future<T2<UUID, Set<UUID>>> fut =
compute.executeAsync(TestTask.class.getName(), null);
+ boolean cancelled = (i % 3 == 0) &&
fut.cancel(true);
+
assertEquals((Integer)i, cache.get(threadIdx));
- assertEquals(nodeIds(nodeIdx), fut.get().get2());
+ if (cancelled)
+ assertTrue(fut.isCancelled());
+ else
+ assertEquals(nodeIds(nodeIdx),
fut.get().get2());
}
}
catch (ExecutionException e) {
@@ -527,7 +530,8 @@ public class ComputeTaskTest extends GridCommonAbstractTest
{
}, threadsCnt, "run-task-async");
-
assertTrue(((ClientComputeImpl)client.compute()).activeTaskFutures().isEmpty());
+ assertTrue(GridTestUtils.waitForCondition(
+ () ->
((ClientComputeImpl)client.compute()).activeTaskFutures().isEmpty(), TIMEOUT));
}
}
