ignite-28305 review

Author: Egor Kuts

modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java

@@ -519,6 +519,9 @@ public static class Replicator {
 
         /** Replica is absent on the node and the node is not in assignments for this replica. */
         public static final int REPLICA_ABSENT_ERR = REPLICATOR_ERR_GROUP.registerErrorCode((short) 11);
+
+        /** Node is overloaded: in-flight partition operation byte limit reached. */
+        public static final int REPLICA_OVERLOADED_ERR = REPLICATOR_ERR_GROUP.registerErrorCode((short) 12);
     }
 
     /** Storage error group. */

modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java

@@ -47,7 +47,6 @@
 import org.apache.ignite.internal.sql.engine.QueryProcessor;
 import org.apache.ignite.internal.table.IgniteTablesInternal;
 import org.apache.ignite.internal.tx.TxManager;
-import org.apache.ignite.internal.util.PartitionOperationInFlightLimiter;
 import org.apache.ignite.network.NetworkAddress;
 import org.jetbrains.annotations.Nullable;
 import org.junit.jupiter.api.TestInfo;
@@ -145,7 +144,6 @@ ClientHandlerModule start(TestInfo testInfo) {
                 EventLog.NOOP,
                 new TestLowWatermark(),
                 Runnable::run,
-                new PartitionOperationInFlightLimiter(0),
                 () -> true
         );
 

modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java

@@ -74,7 +74,6 @@
 import org.apache.ignite.internal.table.IgniteTablesInternal;
 import org.apache.ignite.internal.tx.TxManager;
 import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.apache.ignite.internal.util.PartitionOperationInFlightLimiter;
 import org.apache.ignite.lang.IgniteException;
 import org.jetbrains.annotations.Nullable;
 import org.jetbrains.annotations.TestOnly;
@@ -164,8 +163,6 @@ public class ClientHandlerModule implements IgniteComponent, PlatformComputeTran
 
     private final Executor partitionOperationsExecutor;
 
-    private final PartitionOperationInFlightLimiter partitionOperationInFlightLimiter;
-
     private final ConcurrentHashMap<String, CompletableFuture<PlatformComputeConnection>> computeExecutors = new ConcurrentHashMap<>();
 
     @TestOnly
@@ -189,7 +186,6 @@ public class ClientHandlerModule implements IgniteComponent, PlatformComputeTran
      * @param eventLog Event log.
      * @param lowWatermark Low watermark.
      * @param partitionOperationsExecutor Executor for a partition operation.
-     * @param partitionOperationInFlightLimiter In-flight limiter for partition operations.
      * @param ddlBatchingSuggestionEnabled Boolean supplier indicates whether the suggestion related DDL batching is enabled.
      */
     public ClientHandlerModule(
@@ -211,7 +207,6 @@ public ClientHandlerModule(
             EventLog eventLog,
             LowWatermark lowWatermark,
             Executor partitionOperationsExecutor,
-            PartitionOperationInFlightLimiter partitionOperationInFlightLimiter,
             Supplier<Boolean> ddlBatchingSuggestionEnabled
     ) {
         assert igniteTables != null;
@@ -257,7 +252,6 @@ public ClientHandlerModule(
         this.clientConnectorConfiguration = clientConnectorConfiguration;
         this.ddlBatchingSuggestionEnabled = ddlBatchingSuggestionEnabled;
         this.partitionOperationsExecutor = partitionOperationsExecutor;
-        this.partitionOperationInFlightLimiter = partitionOperationInFlightLimiter;
     }
 
     /** {@inheritDoc} */
@@ -477,7 +471,6 @@ private ClientInboundMessageHandler createInboundMessageHandler(
                 connectionId,
                 primaryReplicaTracker,
                 partitionOperationsExecutor,
-                partitionOperationInFlightLimiter,
                 SUPPORTED_FEATURES,
                 Map.of(),
                 computeExecutors::remove,

modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java

@@ -153,7 +153,6 @@
 import org.apache.ignite.internal.jdbc.proto.JdbcQueryCursorHandler;
 import org.apache.ignite.internal.lang.IgniteExceptionMapperUtil;
 import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
-import org.apache.ignite.internal.lang.ReplicaOverloadedException;
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.logger.Loggers;
 import org.apache.ignite.internal.network.ClusterService;
@@ -181,7 +180,6 @@
 import org.apache.ignite.internal.tx.TransactionKilledException;
 import org.apache.ignite.internal.tx.TxManager;
 import org.apache.ignite.internal.util.ExceptionUtils;
-import org.apache.ignite.internal.util.PartitionOperationInFlightLimiter;
 import org.apache.ignite.lang.CancelHandle;
 import org.apache.ignite.lang.ErrorGroups.Compute;
 import org.apache.ignite.lang.ErrorGroups.Sql;
@@ -280,8 +278,6 @@ public class ClientInboundMessageHandler
 
     private final Executor partitionOperationsExecutor;
 
-    private final PartitionOperationInFlightLimiter partitionOperationInFlightLimiter;
-
     private final BitSet features;
 
     private final Map<HandshakeExtension, Object> extensions;
@@ -313,7 +309,6 @@ public class ClientInboundMessageHandler
      * @param connectionId Connection ID.
      * @param primaryReplicaTracker Primary replica tracker.
      * @param partitionOperationsExecutor Partition operations executor.
-     * @param partitionOperationInFlightLimiter In-flight limiter for partition operations.
      * @param features Features.
      * @param extensions Extensions.
      * @param eventLog Event log.
@@ -335,7 +330,6 @@ public ClientInboundMessageHandler(
             long connectionId,
             ClientPrimaryReplicaTracker primaryReplicaTracker,
             Executor partitionOperationsExecutor,
-            PartitionOperationInFlightLimiter partitionOperationInFlightLimiter,
             BitSet features,
             Map<HandshakeExtension, Object> extensions,
             Function<String, CompletableFuture<PlatformComputeConnection>> computeConnectionFunc,
@@ -379,7 +373,6 @@ public ClientInboundMessageHandler(
         this.eventLog = eventLog;
         this.primaryReplicaTracker = primaryReplicaTracker;
         this.partitionOperationsExecutor = partitionOperationsExecutor;
-        this.partitionOperationInFlightLimiter = partitionOperationInFlightLimiter;
         this.handshakeEventLoopSwitcher = handshakeEventLoopSwitcher;
 
         jdbcQueryCursorHandler = new JdbcQueryCursorHandlerImpl(resources);
@@ -427,7 +420,6 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
         ByteBuf byteBuf = (ByteBuf) msg;
-
         // Each inbound handler in a pipeline has to release the received messages.
         var unpacker = new ClientMessageUnpacker(byteBuf);
 
@@ -889,28 +881,18 @@ private void processOperation(ChannelHandlerContext ctx, ClientMessageUnpacker i
             if (ClientOp.isPartitionOperation(opCode)) {
                 long requestId0 = requestId;
                 int opCode0 = opCode;
-                if (!partitionOperationInFlightLimiter.tryAcquire()) {
-                    in.close();
+                partitionOperationsExecutor.execute(() -> {
+                    try {
+                        processOperationInternal(ctx, in, requestId0, opCode0, guard);
+                    } catch (Throwable t) {
+                        in.close();
 
-                    writeError(requestId0, opCode0, new ReplicaOverloadedException(), ctx, false, guard);
+                        writeError(requestId0, opCode0, t, ctx, false, guard);
 
-                    metrics.requestsFailedIncrement();
-                } else {
-                    partitionOperationsExecutor.execute(() -> {
-                        try {
-                            processOperationInternal(ctx, in, requestId0, opCode0, guard);
-                        } catch (Throwable t) {
-                            in.close();
-
-                            writeError(requestId0, opCode0, t, ctx, false, guard);
-
-                            metrics.requestsFailedIncrement();
-                            metrics.requestsActiveDecrement();
-                        } finally {
-                            partitionOperationInFlightLimiter.release();
-                        }
-                    });
-                }
+                        metrics.requestsFailedIncrement();
+                        metrics.requestsActiveDecrement();
+                    }
+                });
             } else {
                 processOperationInternal(ctx, in, requestId, opCode, guard);
             }

modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java

@@ -65,7 +65,6 @@
 import org.apache.ignite.internal.schema.AlwaysSyncedSchemaSyncService;
 import org.apache.ignite.internal.security.authentication.AuthenticationManager;
 import org.apache.ignite.internal.table.IgniteTablesInternal;
-import org.apache.ignite.internal.util.PartitionOperationInFlightLimiter;
 import org.apache.ignite.lang.IgniteException;
 import org.jetbrains.annotations.Nullable;
 
@@ -273,7 +272,6 @@ protected void initChannel(Channel ch) {
                                                 new TestLowWatermark()
                                         ),
                                         Runnable::run,
-                                        new PartitionOperationInFlightLimiter(0),
                                         features,
                                         randomExtensions(),
                                         unused -> null,

modules/client/src/test/java/org/apache/ignite/client/TestServer.java

@@ -79,7 +79,6 @@
 import org.apache.ignite.internal.security.authentication.AuthenticationManagerImpl;
 import org.apache.ignite.internal.security.configuration.SecurityConfiguration;
 import org.apache.ignite.internal.table.IgniteTablesInternal;
-import org.apache.ignite.internal.util.PartitionOperationInFlightLimiter;
 import org.apache.ignite.network.NetworkAddress;
 import org.jetbrains.annotations.Nullable;
 import org.mockito.Mockito;
@@ -291,7 +290,6 @@ public void log(String type, Supplier<Event> eventProvider) {
                         EventLog.NOOP,
                         new TestLowWatermark(),
                         Runnable::run,
-                        new PartitionOperationInFlightLimiter(0),
                         () -> true
                 );
 

modules/core/src/main/java/org/apache/ignite/internal/lang/ReplicaOverloadedException.java

@@ -17,17 +17,17 @@
 
 package org.apache.ignite.internal.lang;
 
-import static org.apache.ignite.lang.ErrorGroups.Replicator.GROUP_OVERLOADED_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Replicator.REPLICA_OVERLOADED_ERR;
 
 /**
- * Thrown when the node has reached the maximum number of in-flight partition operations
- * ({@code replication.maxInFlightPartitionOperationsPerCore}) and cannot accept new requests.
+ * Thrown when the node has reached the in-flight partition operation byte limit
+ * ({@code replication.partitionOperationHeapUsagePercent}) and cannot accept new requests.
  */
 public class ReplicaOverloadedException extends IgniteInternalException {
     private static final long serialVersionUID = -6023736883539658779L;
 
     /** Constructor. */
     public ReplicaOverloadedException() {
-        super(GROUP_OVERLOADED_ERR, "Node is overloaded: max in-flight partition operations limit reached.");
+        super(REPLICA_OVERLOADED_ERR, "Node is overloaded: in-flight partition operation byte limit reached.");
     }
 }