IGNITE-28734 Process control.sh messages in management pool - Fixes #13192.

Signed-off-by: Aleksey Plekhanov <plehanov.alex@gmail.com>

Author: alex-plekhanov

modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite.java

@@ -34,6 +34,7 @@
 import org.apache.ignite.util.GridCommandHandlerIndexingTest;
 import org.apache.ignite.util.GridCommandHandlerIndexingWithSSLTest;
 import org.apache.ignite.util.GridCommandHandlerLegacyClientTest;
+import org.apache.ignite.util.GridCommandHandlerManagementPoolTest;
 import org.apache.ignite.util.GridCommandHandlerMetadataTest;
 import org.apache.ignite.util.GridCommandHandlerSslTest;
 import org.apache.ignite.util.GridCommandHandlerTest;
@@ -80,7 +81,8 @@
     BaselineEventsRemoteTest.class,
 
     GridCommandHandlerWalTest.class,
-    GridCommandHandlerCheckpointTest.class
+    GridCommandHandlerCheckpointTest.class,
+    GridCommandHandlerManagementPoolTest.class,
 })
 public class IgniteControlUtilityTestSuite {
 }

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerManagementPoolTest.java

@@ -0,0 +1,94 @@
+/*
+ * 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.util;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.annotations.QuerySqlFunction;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.Test;
+
+import static org.apache.ignite.Ignition.startClient;
+import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_OK;
+
+/**
+ * Tests management pool usage for management tasks.
+ */
+public class GridCommandHandlerManagementPoolTest extends GridCommandHandlerClusterPerMethodAbstractTest {
+    /** */
+    private static final long TIMEOUT = 10_000L;
+
+    /** */
+    @Test
+    public void testManagementTasksWorksWhenClientPoolBlocked() throws Exception {
+        Ignite ignite = startGrid(0);
+
+        assertEquals(EXIT_CODE_OK, execute("--set-state", ClusterState.ACTIVE.name()));
+
+        ignite.createCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+            .setSqlSchema(QueryUtils.DFLT_SCHEMA)
+            .setSqlFunctionClasses(TestSqlFunctions.class)
+            .setIndexedTypes(Integer.class, String.class)
+        );
+
+        TestSqlFunctions.latch = new CountDownLatch(1);
+
+        try (IgniteClient client = startClient(new ClientConfiguration().setAddresses("127.0.0.1:10800"))) {
+            // Block client pool by SQL queries.
+            IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(
+                () -> client.query(new SqlFieldsQuery("SELECT wait_latch()")).getAll(),
+                ClientConnectorConfiguration.DFLT_THREAD_POOL_SIZE, "client-thread");
+
+            // Check that management tasks still can be processed.
+            assertEquals(EXIT_CODE_OK, execute("--state")); // Native command.
+            assertEquals(EXIT_CODE_OK, execute("--checkpoint")); // Multi-node task command.
+
+            TestSqlFunctions.latch.countDown();
+
+            fut.get(TIMEOUT, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    /** */
+    public static class TestSqlFunctions {
+        /** */
+        private static CountDownLatch latch;
+
+        /** */
+        @QuerySqlFunction
+        public static boolean wait_latch() {
+            try {
+                latch.await(TIMEOUT, TimeUnit.MILLISECONDS);
+            }
+            catch (InterruptedException ignored) {
+                return false;
+            }
+
+            return true;
+        }
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerAbstractConnectionContext.java

@@ -32,6 +32,7 @@
 import org.apache.ignite.plugin.security.SecurityCredentials;
 import org.jetbrains.annotations.Nullable;
 
+import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.MANAGEMENT_CLIENT_ATTR;
 import static org.apache.ignite.plugin.security.SecuritySubjectType.REMOTE_CLIENT;
 
 /**
@@ -56,6 +57,9 @@ public abstract class ClientListenerAbstractConnectionContext implements ClientL
     /** User attributes. */
     protected Map<String, String> userAttrs;
 
+    /** If client is management. */
+    private volatile Boolean managementClient;
+
     /**
      * Describes the client connection:
      * - thin cli: "cli:host:port@user_name"
@@ -138,11 +142,13 @@ protected void authenticate(GridNioSession ses, String user, String pwd) throws
     }
 
     /** */
-    protected void initClientDescriptor(String prefix) {
+    protected void initClientContext(String prefix) {
         clientDesc = prefix + ":" + ses.remoteAddress().getHostString() + ":" + ses.remoteAddress().getPort();
 
         if (secCtx != null)
             clientDesc += "@" + secCtx.subject().login();
+
+        managementClient = Boolean.parseBoolean(attributes().get(MANAGEMENT_CLIENT_ATTR));
     }
 
     /**
@@ -196,4 +202,9 @@ public int nextTxId() {
     @Override public Map<String, String> attributes() {
         return F.isEmpty(userAttrs) ? Collections.emptyMap() : Collections.unmodifiableMap(userAttrs);
     }
+
+    /** {@inheritDoc} */
+    @Override public boolean managementClient() {
+        return managementClient;
+    }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerConnectionContext.java

@@ -24,8 +24,6 @@
 import org.apache.ignite.internal.util.nio.GridNioSession;
 import org.jetbrains.annotations.Nullable;
 
-import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.MANAGEMENT_CLIENT_ATTR;
-
 /**
  * SQL listener connection context.
  */
@@ -94,7 +92,5 @@ void initializeFromHandshake(GridNioSession ses, ClientListenerProtocolVersion v
     /**
      * @return {@code True} if client is management.
      */
-    default boolean managementClient() {
-        return Boolean.parseBoolean(attributes().get(MANAGEMENT_CLIENT_ATTR));
-    }
+    boolean managementClient();
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java

@@ -59,6 +59,8 @@
 import org.apache.ignite.internal.util.nio.ssl.GridNioSslFilter;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.internal.util.worker.GridWorker;
+import org.apache.ignite.internal.util.worker.GridWorkerPool;
 import org.apache.ignite.lang.IgniteBiInClosure;
 import org.apache.ignite.metric.MetricRegistry;
 import org.apache.ignite.mxbean.ClientProcessorMXBean;
@@ -122,6 +124,9 @@ public class ClientListenerProcessor extends GridProcessorAdapter {
     /** Executor service. */
     private ExecutorService execSvc;
 
+    /** Management pool. */
+    private GridWorkerPool mgmtPool;
+
     /** Thin client distributed configuration. */
     private DistributedThinClientConfiguration distrThinCfg;
 
@@ -161,6 +166,7 @@ public ClientListenerProcessor(GridKernalContext ctx) {
                 }
 
                 execSvc = ctx.pools().getThinClientExecutorService();
+                mgmtPool = new GridWorkerPool(ctx.pools().getManagementExecutorService(), log);
 
                 Exception lastErr = null;
 
@@ -435,10 +441,43 @@ private void unregisterMBean() {
                     else {
                         connCtx.handler().registerRequest(reqId, cmdType);
 
-                        super.onMessageReceived(ses, msg);
+                        onMessageReceived(ses, connCtx, msg);
                     }
                 }
                 else
+                    onMessageReceived(ses, connCtx, msg);
+            }
+
+            /** */
+            private void onMessageReceived(
+                GridNioSession ses,
+                @Nullable ClientListenerConnectionContext connCtx,
+                Object msg
+            ) throws IgniteCheckedException {
+                if (connCtx == null) {
+                    // Process handshake in NIO thread.
+                    try {
+                        proceedMessageReceived(ses, msg);
+                    }
+                    catch (IgniteCheckedException e) {
+                        handleException(ses, e);
+                    }
+                }
+                else if (connCtx.managementClient()) {
+                    // Process management messages in management pool.
+                    mgmtPool.execute(
+                        new GridWorker(ctx.igniteInstanceName(), "management-message-received-notify", log) {
+                            @Override protected void body() {
+                                try {
+                                    proceedMessageReceived(ses, msg);
+                                }
+                                catch (IgniteCheckedException e) {
+                                    handleException(ses, e);
+                                }
+                            }
+                        });
+                }
+                else // Process regular messages in client-listener pool.
                     super.onMessageReceived(ses, msg);
             }
         };
@@ -488,6 +527,10 @@ private void unregisterMBean() {
 
             execSvc = null;
 
+            mgmtPool.join(cancel);
+
+            mgmtPool = null;
+
             if (!U.IGNITE_MBEANS_DISABLED)
                 unregisterMBean();
 

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/jdbc/JdbcConnectionContext.java

@@ -267,7 +267,7 @@ public JdbcConnectionContext(GridKernalContext ctx, GridNioSession ses, GridSpin
 
         protoCtx = new JdbcProtocolContext(ver, features, true);
 
-        initClientDescriptor("jdbc-thin");
+        initClientContext("jdbc-thin");
 
         parser = new JdbcMessageParser(ctx, protoCtx);
 

modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/odbc/OdbcConnectionContext.java

@@ -198,7 +198,7 @@ public OdbcConnectionContext(GridKernalContext ctx, GridNioSession ses, GridSpin
             }
         };
 
-        initClientDescriptor("odbc");
+        initClientContext("odbc");
 
         handler = new OdbcRequestHandler(ctx, busyLock, snd, maxCursors, distributedJoins, enforceJoinOrder,
             replicatedOnly, collocated, skipReducerOnUpdate, qryEngine, ver, this);

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java

@@ -226,7 +226,7 @@ public ClientProtocolContext currentProtocolContext() {
 
         authenticate(ses, user, pwd);
 
-        initClientDescriptor("cli");
+        initClientContext("cli");
 
         handler = new ClientRequestHandler(this, currentProtocolContext);
         parser = new ClientMessageParser(this, currentProtocolContext);

modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioAsyncNotifyFilter.java

@@ -139,7 +139,7 @@ public GridNioAsyncNotifyFilter(String igniteInstanceName, Executor exec, Ignite
      * @param ses Session.
      * @param ex Exception.
      */
-    private void handleException(GridNioSession ses, IgniteCheckedException ex) {
+    protected void handleException(GridNioSession ses, IgniteCheckedException ex) {
         try {
             proceedExceptionCaught(ses, ex);
         }