GG-37994 Task event can record attributes.

modules/core/src/test/java/org/apache/ignite/session/GridSessionSetTaskAttributeSelfTest.java

@@ -19,27 +19,45 @@
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.compute.ComputeJob;
 import org.apache.ignite.compute.ComputeJobAdapter;
+import org.apache.ignite.compute.ComputeJobContext;
 import org.apache.ignite.compute.ComputeJobResult;
 import org.apache.ignite.compute.ComputeJobResultPolicy;
 import org.apache.ignite.compute.ComputeTaskFuture;
 import org.apache.ignite.compute.ComputeTaskSession;
 import org.apache.ignite.compute.ComputeTaskSessionFullSupport;
 import org.apache.ignite.compute.ComputeTaskSplitAdapter;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.TaskEvent;
+import org.apache.ignite.failure.StopNodeFailureHandler;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.compute.ComputeTaskWithWithoutFullSupportTest;
 import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.lang.IgniteRunnable;
+import org.apache.ignite.resources.JobContextResource;
 import org.apache.ignite.resources.LoggerResource;
+import org.apache.ignite.resources.TaskSessionAttributes;
 import org.apache.ignite.resources.TaskSessionResource;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.apache.ignite.testframework.junits.common.GridCommonTest;
 import org.junit.Test;
 
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.events.EventType.EVTS_TASK_EXECUTION;
+import static org.apache.ignite.events.EventType.EVT_TASK_STARTED;
+
 /**
  *
  */
@@ -51,11 +69,22 @@ public class GridSessionSetTaskAttributeSelfTest extends GridCommonAbstractTest
     /** */
     public static final int EXEC_COUNT = 5;
 
+    /** */
+    private static final String INT_PARAM_NAME = "customIntParameter";
+    /** */
+    private static final String TXT_PARAM_NAME = "customTextParameter";
+
     /** */
     public GridSessionSetTaskAttributeSelfTest() {
         super(true);
     }
 
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        return super.getConfiguration(igniteInstanceName)
+                .setIncludeEventTypes(EVTS_TASK_EXECUTION);
+    }
+
     /**
      * @throws Exception if failed.
      */
@@ -101,6 +130,41 @@ public void testMultiThreaded() throws Exception {
             fail();
     }
 
+    /**
+     * Check parameters propagation from session to events
+     * @throws Exception
+     */
+    @Test
+    public void testAttributesToEventsPropagation() throws Exception {
+        final int intParam = 1;
+        final String textParam = "text";
+
+        IgniteEx n = startGrid(0);
+
+        n.cluster().state(ACTIVE);
+
+        IgnitePredicate<TaskEvent> lsnr = evt -> {
+            System.out.println("Received task event [evt=" + evt.name() + ", taskName=" + evt.taskName() + ", taskAttributes=" + evt.attributes() + ']');
+
+            if (evt.type() != EVT_TASK_STARTED && evt.attributes().isEmpty()) {
+                assertEquals(intParam, evt.attributes().get(INT_PARAM_NAME));
+                assertEquals(textParam, evt.attributes().get(TXT_PARAM_NAME));
+            }
+
+            return true;
+        };
+
+        n.events().localListen(lsnr, EVTS_TASK_EXECUTION);
+
+        // Generate task events.
+        IgniteFuture<Void> taskFut0 = n.compute().runAsync(new CallableWithSessionAttributes(intParam, textParam));
+
+        taskFut0.get();
+
+        // Unsubscribe local task event listener.
+        n.events().stopLocalListen(lsnr);
+    }
+
     /**
      * @param num Number.
      */
@@ -114,6 +178,33 @@ private void checkTask(int num) {
         assert (Integer)res == SPLIT_COUNT : "Invalid result [num=" + num + ", fut=" + fut + ']';
     }
 
+    /**
+     *
+     */
+    @ComputeTaskSessionFullSupport
+    private static class CallableWithSessionAttributes implements IgniteRunnable {
+
+        @TaskSessionResource
+        private ComputeTaskSession ses;
+
+        private int numericParameter;
+        private String textParameter;
+
+        public CallableWithSessionAttributes(int numericParameter, String textParameter) {
+            this.numericParameter = numericParameter;
+            this.textParameter = textParameter;
+        }
+
+        @Override
+        public void run() {
+            if (numericParameter != 0)
+                ses.setAttribute(INT_PARAM_NAME, numericParameter);
+
+            if (textParameter != null)
+                ses.setAttribute(TXT_PARAM_NAME, textParameter);
+        }
+    }
+
     /**
      *
      */