IGNITE-13827 Java thin client: Fixed hang on ComputeTask returning un…

modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientComputeImpl.java

@@ -358,13 +358,20 @@ private <T> void writeExecuteTaskRequest(
         Exception err
     ) {
         if (op == ClientOperation.COMPUTE_TASK_FINISHED) {
-            Object res = payload == null ? null : utils.readObject(BinaryByteBufferInputStream.create(payload), false);
-
             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);
+                if (err == null) {
+                    try {
+                        Object res = payload == null ? null :
+                            utils.readObject(BinaryByteBufferInputStream.create(payload), false);
+
+                        task.fut.onDone(res);
+                    }
+                    catch (Throwable e) {
+                        task.fut.onDone(e);
+                    }
+                }
                 else
                     task.fut.onDone(err);
 

modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientChannel.java

@@ -378,19 +378,22 @@ private void processNextMessage(ByteBuffer buf) throws ClientProtocolError, Clie
         int hdrSize = dataInput.position();
         int msgSize = buf.limit();
 
-        ByteBuffer res = null;
-        Exception err = null;
+        ByteBuffer res;
+        Exception err;
 
         if (status == 0) {
-            if (msgSize > hdrSize)
-                res = buf;
+            err = null;
+            res = msgSize > hdrSize ? buf : null;
         }
-        else if (status == ClientStatus.SECURITY_VIOLATION)
+        else if (status == ClientStatus.SECURITY_VIOLATION) {
             err = new ClientAuthorizationException();
+            res = null;
+        }
         else {
             String errMsg = ClientUtils.createBinaryReader(null, dataInput).readString();
 
             err = new ClientServerError(errMsg, status, resId);
+            res = null;
         }
 
         if (notificationOp == null) { // Respone received.
@@ -402,8 +405,12 @@ else if (status == ClientStatus.SECURITY_VIOLATION)
             pendingReq.onDone(res, err);
         }
         else { // Notification received.
-            for (NotificationListener lsnr : notificationLsnrs)
-                lsnr.acceptNotification(this, notificationOp, resId, res, err);
+            ClientOperation notificationOp0 = notificationOp;
+
+            asyncContinuationExecutor.execute(() -> {
+                for (NotificationListener lsnr : notificationLsnrs)
+                    lsnr.acceptNotification(this, notificationOp0, resId, res, err);
+            });
         }
     }
 

modules/core/src/test/java/org/apache/ignite/internal/client/thin/ComputeTaskTest.java

@@ -45,6 +45,7 @@
 import org.apache.ignite.client.ClientException;
 import org.apache.ignite.client.IgniteClient;
 import org.apache.ignite.client.IgniteClientFuture;
+import org.apache.ignite.client.Person;
 import org.apache.ignite.compute.ComputeJobResult;
 import org.apache.ignite.compute.ComputeTaskName;
 import org.apache.ignite.configuration.ClientConnectorConfiguration;
@@ -119,6 +120,18 @@ public void testExecuteTaskByClassName() throws Exception {
         }
     }
 
+    /**
+     * Test that custom user type can be returned by task.
+     */
+    @Test
+    public void testExecuteWithCustomUserType() throws Exception {
+        try (IgniteClient client = startClient(0)) {
+            Person val = client.compute().execute(TestTaskCustomType.class.getName(), "person");
+
+            assertEquals("person", val.getName());
+        }
+    }
+
     /**
      *
      */

modules/core/src/test/java/org/apache/ignite/internal/client/thin/TestTaskCustomType.java

@@ -0,0 +1,47 @@
+/*
+ * 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.client.thin;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.client.Person;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobAdapter;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ * Compute task which returns custom user type.
+ */
+public class TestTaskCustomType extends ComputeTaskAdapter<String, Person> {
+    /** {@inheritDoc} */
+    @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, String arg) {
+        return F.asMap(new ComputeJobAdapter() {
+            @Override public Object execute() {
+                return arg;
+            }
+        }, F.first(subgrid));
+    }
+
+    /** {@inheritDoc} */
+    @Override public Person reduce(List<ComputeJobResult> results) {
+        return new Person(0, F.first(results).getData());
+    }
+}