Merge branch 'main' into threadpool-merge-scheduler

Author: albertzaharovits

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

@@ -75,13 +75,15 @@ public void addTracingHandler(ChunkHandler chunkHandler) {
 
     @Override
     public void next() {
-        assert closing == false : "cannot request next chunk on closing stream";
         assert handler != null : "handler must be set before requesting next chunk";
         requestContext = threadContext.newStoredContext();
         channel.eventLoop().submit(() -> {
             activityTracker.startActivity();
             requested = true;
             try {
+                if (closing) {
+                    return;
+                }
                 if (buf == null) {
                     channel.read();
                 } else {

muted-tests.yml

@@ -178,9 +178,6 @@ tests:
   issue: https://github.com/elastic/elasticsearch/issues/118374
 - class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
   issue: https://github.com/elastic/elasticsearch/issues/118238
-- class: org.elasticsearch.reservedstate.service.FileSettingsServiceTests
-  method: testInvalidJSON
-  issue: https://github.com/elastic/elasticsearch/issues/116521
 - class: org.elasticsearch.xpack.ccr.rest.ShardChangesRestIT
   method: testShardChangesNoOperation
   issue: https://github.com/elastic/elasticsearch/issues/118800
@@ -244,11 +241,17 @@ tests:
 - class: org.elasticsearch.search.ccs.CrossClusterIT
   method: testCancel
   issue: https://github.com/elastic/elasticsearch/issues/108061
-- class: org.elasticsearch.http.netty4.Netty4IncrementalRequestHandlingIT
-  method: testOversizedChunkedEncoding
-  issue: https://github.com/elastic/elasticsearch/issues/120444
 - class: org.elasticsearch.xpack.logsdb.seqno.RetentionLeaseRestIT
   issue: https://github.com/elastic/elasticsearch/issues/120434
+- class: org.elasticsearch.entitlement.qa.EntitlementsAllowedIT
+  method: testCheckActionWithPolicyPass {pathPrefix=allowed actionName=create_ldap_cert_store}
+  issue: https://github.com/elastic/elasticsearch/issues/120422
+- class: org.elasticsearch.entitlement.qa.EntitlementsAllowedIT
+  method: testCheckActionWithPolicyPass {pathPrefix=allowed_nonmodular actionName=create_ldap_cert_store}
+  issue: https://github.com/elastic/elasticsearch/issues/120423
+- class: org.elasticsearch.reservedstate.service.FileSettingsServiceTests
+  method: testInvalidJSON
+  issue: https://github.com/elastic/elasticsearch/issues/120482
 
 # Examples:
 #

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -156,7 +156,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion ELASTIC_INFERENCE_SERVICE_UNIFIED_CHAT_COMPLETIONS_INTEGRATION = def(8_822_00_0);
     public static final TransportVersion KQL_QUERY_TECH_PREVIEW = def(8_823_00_0);
     public static final TransportVersion ESQL_PROFILE_ROWS_PROCESSED = def(8_824_00_0);
-    public static final TransportVersion BYTE_SIZE_VALUE_ALWAYS_USES_BYTES = def(8_825_00_0);
+    public static final TransportVersion BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1 = def(8_825_00_0);
+    public static final TransportVersion REVERT_BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1 = def(8_826_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/common/unit/ByteSizeValue.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.common.unit;
 
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -24,7 +25,8 @@
 import java.util.Locale;
 import java.util.Objects;
 
-import static org.elasticsearch.TransportVersions.BYTE_SIZE_VALUE_ALWAYS_USES_BYTES;
+import static org.elasticsearch.TransportVersions.BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1;
+import static org.elasticsearch.TransportVersions.REVERT_BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1;
 import static org.elasticsearch.common.unit.ByteSizeUnit.BYTES;
 import static org.elasticsearch.common.unit.ByteSizeUnit.GB;
 import static org.elasticsearch.common.unit.ByteSizeUnit.KB;
@@ -111,7 +113,8 @@ static ByteSizeValue newByteSizeValue(long sizeInBytes, ByteSizeUnit desiredUnit
     public static ByteSizeValue readFrom(StreamInput in) throws IOException {
         long size = in.readZLong();
         ByteSizeUnit unit = ByteSizeUnit.readFrom(in);
-        if (in.getTransportVersion().onOrAfter(BYTE_SIZE_VALUE_ALWAYS_USES_BYTES)) {
+        TransportVersion tv = in.getTransportVersion();
+        if (tv.onOrAfter(BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1) && tv.before(REVERT_BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1)) {
             return newByteSizeValue(size, unit);
         } else {
             return of(size, unit);
@@ -120,7 +123,8 @@ public static ByteSizeValue readFrom(StreamInput in) throws IOException {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        if (out.getTransportVersion().onOrAfter(BYTE_SIZE_VALUE_ALWAYS_USES_BYTES)) {
+        TransportVersion tv = out.getTransportVersion();
+        if (tv.onOrAfter(BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1) && tv.before(REVERT_BYTE_SIZE_VALUE_ALWAYS_USES_BYTES_1)) {
             out.writeZLong(sizeInBytes);
         } else {
             out.writeZLong(Math.divideExact(sizeInBytes, desiredUnit.toBytes(1)));

server/src/main/java/org/elasticsearch/index/CompositeIndexEventListener.java

@@ -28,6 +28,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.core.Strings.format;
 
@@ -349,4 +350,16 @@ public void afterFilesRestoredFromRepository(IndexShard indexShard) {
             }
         }
     }
+
+    @Override
+    public void onAcquirePrimaryOperationPermit(IndexShard indexShard, Supplier<ActionListener<Void>> onPermitAcquiredListenerSupplier) {
+        for (IndexEventListener listener : listeners) {
+            try {
+                listener.onAcquirePrimaryOperationPermit(indexShard, onPermitAcquiredListenerSupplier);
+            } catch (Exception e) {
+                logger.warn(() -> "[" + indexShard.shardId() + "] failed to invoke the listener on acquiring a primary permit", e);
+                throw e;
+            }
+        }
+    }
 }

server/src/main/java/org/elasticsearch/index/shard/IndexEventListener.java

@@ -17,6 +17,8 @@
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.indices.cluster.IndicesClusterStateService.AllocatedIndices.IndexRemovalReason;
 
+import java.util.function.Supplier;
+
 /**
  * An index event listener is the primary extension point for plugins and build-in services
  * to react / listen to per-index and per-shard events. These listeners are registered per-index
@@ -190,4 +192,14 @@ default void afterIndexShardRecovery(IndexShard indexShard, ActionListener<Void>
      * @param indexShard the shard that is recovering
      */
     default void afterFilesRestoredFromRepository(IndexShard indexShard) {}
+
+    /**
+     * Called when a single primary permit is acquired for the given shard (see
+     * {@link IndexShard#acquirePrimaryOperationPermit(ActionListener, java.util.concurrent.Executor)}).
+     *
+     * @param indexShard the shard of which a primary permit is requested
+     * @param onPermitAcquiredListenerSupplier call this immediately to get a listener when the permit is acquired. The listener must be
+     *                                         completed in order for the permit to be given to the acquiring operation.
+     */
+    default void onAcquirePrimaryOperationPermit(IndexShard indexShard, Supplier<ActionListener<Void>> onPermitAcquiredListenerSupplier) {}
 }

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.RefCountingListener;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.action.support.replication.PendingReplicationActions;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
@@ -189,7 +190,6 @@
 import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.index.seqno.RetentionLeaseActions.RETAIN_ALL;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
-import static org.elasticsearch.index.shard.IndexShard.PrimaryPermitCheck.CHECK_PRIMARY_MODE;
 
 public class IndexShard extends AbstractIndexShardComponent implements IndicesClusterStateService.Shard {
 
@@ -779,28 +779,10 @@ public void relocated(
         final String targetAllocationId,
         final BiConsumer<ReplicationTracker.PrimaryContext, ActionListener<Void>> consumer,
         final ActionListener<Void> listener
-    ) throws IllegalIndexShardStateException, IllegalStateException {
-        relocated(targetNodeId, targetAllocationId, consumer, listener, null);
-    }
-
-    /**
-     * Provides an variant of {@link IndexShard#relocated(String, String, BiConsumer, ActionListener, Releasable)} with an option
-     * to relocate the shard under externally acquired primary permits.
-     *
-     * @param acquiredPrimaryPermits if null, waits until all the primary permits are acquired, otherwise it calls the consumer immediately
-     */
-    public void relocated(
-        final String targetNodeId,
-        final String targetAllocationId,
-        final BiConsumer<ReplicationTracker.PrimaryContext, ActionListener<Void>> consumer,
-        final ActionListener<Void> listener,
-        @Nullable final Releasable acquiredPrimaryPermits
     ) throws IllegalIndexShardStateException, IllegalStateException {
         assert shardRouting.primary() : "only primaries can be marked as relocated: " + shardRouting;
-        assert acquiredPrimaryPermits == null || indexShardOperationPermits.getActiveOperationsCount() == OPERATIONS_BLOCKED
-            : "external primary permits are provided but not held by the shard";
         try (Releasable forceRefreshes = refreshListeners.forceRefreshes()) {
-            ActionListener<Releasable> onAcquired = new ActionListener<>() {
+            indexShardOperationPermits.blockOperations(new ActionListener<>() {
                 @Override
                 public void onResponse(Releasable releasable) {
                     boolean success = false;
@@ -878,13 +860,8 @@ public void onFailure(Exception e) {
                         listener.onFailure(e);
                     }
                 }
-            };
-            if (acquiredPrimaryPermits == null) {
-                // Wait on current thread because this execution is wrapped by CancellableThreads and we want to be able to interrupt it
-                indexShardOperationPermits.blockOperations(onAcquired, 30L, TimeUnit.MINUTES, EsExecutors.DIRECT_EXECUTOR_SERVICE);
-            } else {
-                ActionListener.completeWith(onAcquired, () -> acquiredPrimaryPermits);
-            }
+            }, 30L, TimeUnit.MINUTES, EsExecutors.DIRECT_EXECUTOR_SERVICE); // Wait on current thread because this execution is wrapped by
+                                                                            // CancellableThreads and we want to be able to interrupt it
         }
     }
 
@@ -3592,100 +3569,69 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) {
         );
     }
 
-    /**
-     * Check to run before running the primary permit operation
-     */
-    public enum PrimaryPermitCheck {
-        CHECK_PRIMARY_MODE,
-        /**
-         * IMPORTANT: Currently intented to be used only for acquiring primary permits during the recovery of hollow shards.
-         * Don't disable primary mode checks unless you're really sure.
-         */
-        NONE
-    }
-
     /**
      * Acquire a primary operation permit whenever the shard is ready for indexing. If a permit is directly available, the provided
      * ActionListener will be called on the calling thread. During relocation hand-off, permit acquisition can be delayed. The provided
      * ActionListener will then be called using the provided executor.
      */
     public void acquirePrimaryOperationPermit(ActionListener<Releasable> onPermitAcquired, Executor executorOnDelay) {
-        acquirePrimaryOperationPermit(onPermitAcquired, executorOnDelay, false, CHECK_PRIMARY_MODE);
+        acquirePrimaryOperationPermit(onPermitAcquired, executorOnDelay, false);
     }
 
     public void acquirePrimaryOperationPermit(
         ActionListener<Releasable> onPermitAcquired,
         Executor executorOnDelay,
         boolean forceExecution
-    ) {
-        acquirePrimaryOperationPermit(onPermitAcquired, executorOnDelay, forceExecution, CHECK_PRIMARY_MODE);
-    }
-
-    public void acquirePrimaryOperationPermit(
-        ActionListener<Releasable> onPermitAcquired,
-        Executor executorOnDelay,
-        boolean forceExecution,
-        PrimaryPermitCheck primaryPermitCheck
     ) {
         verifyNotClosed();
         assert shardRouting.primary() : "acquirePrimaryOperationPermit should only be called on primary shard: " + shardRouting;
-        indexShardOperationPermits.acquire(
-            wrapPrimaryOperationPermitListener(primaryPermitCheck, onPermitAcquired),
-            executorOnDelay,
-            forceExecution
-        );
+
+        ActionListener<Releasable> onPermitAcquiredWrapped = onPermitAcquired.delegateFailureAndWrap((delegate, releasable) -> {
+            final ActionListener<Releasable> wrappedListener = indexShardOperationPermits.wrapContextPreservingActionListener(
+                delegate,
+                executorOnDelay,
+                forceExecution
+            );
+            try (var listeners = new RefCountingListener(wrappedListener.map(unused -> releasable))) {
+                indexEventListener.onAcquirePrimaryOperationPermit(this, () -> listeners.acquire());
+            }
+        });
+
+        indexShardOperationPermits.acquire(wrapPrimaryOperationPermitListener(onPermitAcquiredWrapped), executorOnDelay, forceExecution);
     }
 
     public boolean isPrimaryMode() {
         assert indexShardOperationPermits.getActiveOperationsCount() != 0 : "must hold permit to check primary mode";
         return replicationTracker.isPrimaryMode();
     }
 
-    public void acquireAllPrimaryOperationsPermits(final ActionListener<Releasable> onPermitAcquired, final TimeValue timeout) {
-        acquireAllPrimaryOperationsPermits(onPermitAcquired, timeout, CHECK_PRIMARY_MODE);
-    }
-
     /**
      * Acquire all primary operation permits. Once all permits are acquired, the provided ActionListener is called.
      * It is the responsibility of the caller to close the {@link Releasable}.
      */
-    public void acquireAllPrimaryOperationsPermits(
-        final ActionListener<Releasable> onPermitAcquired,
-        final TimeValue timeout,
-        final PrimaryPermitCheck primaryPermitCheck
-    ) {
+    public void acquireAllPrimaryOperationsPermits(final ActionListener<Releasable> onPermitAcquired, final TimeValue timeout) {
         verifyNotClosed();
         assert shardRouting.primary() : "acquireAllPrimaryOperationsPermits should only be called on primary shard: " + shardRouting;
 
-        asyncBlockOperations(
-            wrapPrimaryOperationPermitListener(primaryPermitCheck, onPermitAcquired),
-            timeout.duration(),
-            timeout.timeUnit()
-        );
+        asyncBlockOperations(wrapPrimaryOperationPermitListener(onPermitAcquired), timeout.duration(), timeout.timeUnit());
     }
 
     /**
-     * Wraps the action to run on a primary after acquiring permit.
+     * Wraps the action to run on a primary after acquiring permit. This wrapping is used to check if the shard is in primary mode before
+     * executing the action.
      *
-     * @param primaryPermitCheck check to run before the primary mode operation
      * @param listener the listener to wrap
      * @return the wrapped listener
      */
-    private ActionListener<Releasable> wrapPrimaryOperationPermitListener(
-        final PrimaryPermitCheck primaryPermitCheck,
-        final ActionListener<Releasable> listener
-    ) {
-        return switch (primaryPermitCheck) {
-            case CHECK_PRIMARY_MODE -> listener.delegateFailure((l, r) -> {
-                if (isPrimaryMode()) {
-                    l.onResponse(r);
-                } else {
-                    r.close();
-                    l.onFailure(new ShardNotInPrimaryModeException(shardId, state));
-                }
-            });
-            case NONE -> listener;
-        };
+    private ActionListener<Releasable> wrapPrimaryOperationPermitListener(final ActionListener<Releasable> listener) {
+        return listener.delegateFailure((l, r) -> {
+            if (isPrimaryMode()) {
+                l.onResponse(r);
+            } else {
+                r.close();
+                l.onFailure(new ShardNotInPrimaryModeException(shardId, state));
+            }
+        });
     }
 
     private void asyncBlockOperations(ActionListener<Releasable> onPermitAcquired, long timeout, TimeUnit timeUnit) {
@@ -3723,7 +3669,7 @@ public void runUnderPrimaryPermit(final Runnable runnable, final Consumer<Except
                 runnable.run();
             }
         }, onFailure);
-        acquirePrimaryOperationPermit(onPermitAcquired, executorOnDelay, false, CHECK_PRIMARY_MODE);
+        acquirePrimaryOperationPermit(onPermitAcquired, executorOnDelay);
     }
 
     private <E extends Exception> void bumpPrimaryTerm(