Merge branch 'main' into brothermich/ES-10264

Author: BrianRothermich

plugins/examples/custom-suggester/src/main/java/org/elasticsearch/example/customsuggester/CustomSuggester.java

@@ -11,7 +11,7 @@
 
 import org.apache.lucene.search.IndexSearcher;
 import org.apache.lucene.util.CharsRefBuilder;
-import org.elasticsearch.common.text.Text;
+import org.elasticsearch.xcontent.Text;
 import org.elasticsearch.search.suggest.Suggest;
 import org.elasticsearch.search.suggest.Suggester;
 

plugins/examples/custom-suggester/src/main/java/org/elasticsearch/example/customsuggester/CustomSuggestion.java

@@ -11,7 +11,7 @@
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.text.Text;
+import org.elasticsearch.xcontent.Text;
 import org.elasticsearch.search.suggest.Suggest;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.XContentBuilder;

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexStateService.java

@@ -65,6 +65,7 @@
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.IndexReshardService;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.shard.IndexLongFieldRange;
 import org.elasticsearch.index.shard.ShardId;
@@ -352,6 +353,12 @@ static ClusterState addIndexClosedBlocks(
             );
         }
 
+        // Check if index closing conflicts with ongoing resharding
+        Set<Index> reshardingIndices = IndexReshardService.reshardingIndices(currentProjectState, indicesToClose);
+        if (reshardingIndices.isEmpty() == false) {
+            throw new IllegalArgumentException("Cannot close indices that are being resharded: " + reshardingIndices);
+        }
+
         final ClusterBlocks.Builder blocks = ClusterBlocks.builder(currentState.blocks());
 
         for (Index index : indicesToClose) {
@@ -940,6 +947,22 @@ static Tuple<ClusterState, List<IndexResult>> closeRoutingTable(
                     continue;
                 }
 
+                // Check if index closing conflicts with ongoing resharding
+                Set<Index> reshardingIndices = IndexReshardService.reshardingIndices(currentProjectState, Set.of(index));
+                if (reshardingIndices.isEmpty() == false) {
+                    closingResults.put(
+                        result.getKey(),
+                        new IndexResult(
+                            result.getKey(),
+                            new IllegalStateException(
+                                "verification of shards before closing " + index + " succeeded but index is being resharded in the meantime"
+                            )
+                        )
+                    );
+                    logger.debug("verification of shards before closing {} succeeded but index is being resharded in the meantime", index);
+                    continue;
+                }
+
                 blocks.removeIndexBlockWithId(projectId, index.getName(), INDEX_CLOSED_BLOCK_ID);
                 blocks.addIndexBlock(projectId, index.getName(), INDEX_CLOSED_BLOCK);
                 final IndexMetadata.Builder updatedMetadata = IndexMetadata.builder(indexMetadata).state(IndexMetadata.State.CLOSE);

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index;
+
+import org.elasticsearch.cluster.ProjectState;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Resharding is currently only implemented in Serverless. This service only exposes minimal metadata about resharding
+ * needed by other services.
+ */
+public class IndexReshardService {
+    /**
+     * Returns the indices from the provided set that are currently being resharded.
+     */
+    public static Set<Index> reshardingIndices(final ProjectState projectState, final Set<Index> indicesToCheck) {
+        final Set<Index> indices = new HashSet<>();
+        for (Index index : indicesToCheck) {
+            IndexMetadata indexMetadata = projectState.metadata().index(index);
+
+            if (indexMetadata != null && indexMetadata.getReshardingMetadata() != null) {
+                indices.add(index);
+            }
+        }
+        return indices;
+    }
+}

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexStateServiceTests.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexVersion;
@@ -45,6 +46,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -164,6 +166,38 @@ public void testCloseRoutingTableWithSnapshottedIndex() {
         assertThat(updatedState.blocks().hasIndexBlockWithId(projectId, index.getName(), INDEX_CLOSED_BLOCK_ID), is(true));
     }
 
+    public void testCloseRoutingTableWithReshardingIndex() {
+        ClusterState state = stateWithProject("testCloseRoutingTableWithReshardingIndex", projectId);
+
+        String indexName = "resharding-index";
+        ClusterBlock block = MetadataIndexStateService.createIndexClosingBlock();
+        state = addOpenedIndex(projectId, indexName, randomIntBetween(1, 5), randomIntBetween(0, 5), state);
+
+        var updatedMetadata = IndexMetadata.builder(state.metadata().getProject(projectId).index(indexName))
+            .reshardingMetadata(IndexReshardingMetadata.newSplitByMultiple(randomIntBetween(1, 5), 2))
+            .build();
+        state = ClusterState.builder(state)
+            .putProjectMetadata(ProjectMetadata.builder(state.metadata().getProject(projectId)).put(updatedMetadata, true))
+            .build();
+
+        state = ClusterState.builder(state)
+            .blocks(ClusterBlocks.builder().blocks(state.blocks()).addIndexBlock(projectId, indexName, block))
+            .build();
+
+        final Index index = state.metadata().getProject(projectId).index(indexName).getIndex();
+        final Tuple<ClusterState, List<IndexResult>> result = MetadataIndexStateService.closeRoutingTable(
+            state,
+            projectId,
+            Map.of(index, block),
+            Map.of(index, new IndexResult(index)),
+            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY
+        );
+        final ClusterState updatedState = result.v1();
+        assertIsOpened(index.getName(), updatedState, projectId);
+        assertThat(updatedState.blocks().hasIndexBlockWithId(projectId, index.getName(), INDEX_CLOSED_BLOCK_ID), is(true));
+        assertThat(result.v2().get(0).getException(), notNullValue());
+    }
+
     public void testAddIndexClosedBlocks() {
         final ClusterState initialState = stateWithProject("testAddIndexClosedBlocks", projectId);
         {

server/src/test/java/org/elasticsearch/index/IndexReshardServiceTests.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index;
+
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.IndexReshardingMetadata;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class IndexReshardServiceTests extends ESTestCase {
+    public void testReshardingIndices() {
+        var indexMetadata1 = IndexMetadata.builder(randomAlphaOfLength(5)).settings(indexSettings(IndexVersion.current(), 1, 0)).build();
+        var indexMetadata2 = IndexMetadata.builder(indexMetadata1.getIndex().getName() + "2")
+            .settings(indexSettings(IndexVersion.current(), 1, 0))
+            .reshardingMetadata(IndexReshardingMetadata.newSplitByMultiple(1, 2))
+            .build();
+
+        var projectId = randomProjectIdOrDefault();
+
+        var projectMetadataBuilder = ProjectMetadata.builder(projectId);
+        projectMetadataBuilder.put(indexMetadata1, false);
+        projectMetadataBuilder.put(indexMetadata2, false);
+
+        final ClusterState clusterState = ClusterState.builder(new ClusterName("testReshardingIndices"))
+            .putProjectMetadata(projectMetadataBuilder)
+            .build();
+
+        Set<Index> indicesToCheck = new HashSet<>();
+        indicesToCheck.add(indexMetadata1.getIndex());
+        indicesToCheck.add(indexMetadata2.getIndex());
+
+        Set<Index> reshardingIndices = IndexReshardService.reshardingIndices(clusterState.projectState(projectId), indicesToCheck);
+
+        assertEquals(1, reshardingIndices.size());
+        assertTrue(reshardingIndices.contains(indexMetadata2.getIndex()));
+    }
+}

test/framework/src/main/java/org/elasticsearch/indices/CrankyCircuitBreakerService.java

@@ -29,7 +29,7 @@ public class CrankyCircuitBreakerService extends CircuitBreakerService {
      */
     public static final String ERROR_MESSAGE = "cranky breaker";
 
-    private final CircuitBreaker breaker = new CircuitBreaker() {
+    public static final class CrankyCircuitBreaker implements CircuitBreaker {
         private final AtomicLong used = new AtomicLong();
 
         @Override
@@ -82,7 +82,9 @@ public Durability getDurability() {
         public void setLimitAndOverhead(long limit, double overhead) {
 
         }
-    };
+    }
+
+    private final CrankyCircuitBreaker breaker = new CrankyCircuitBreaker();
 
     @Override
     public CircuitBreaker getBreaker(String name) {

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/LocalCircuitBreaker.java

@@ -28,6 +28,7 @@ public final class LocalCircuitBreaker implements CircuitBreaker, Releasable {
     private final long maxOverReservedBytes;
     private long reservedBytes;
     private final AtomicBoolean closed = new AtomicBoolean(false);
+    private volatile Thread activeThread;
 
     public record SizeSettings(long overReservedBytes, long maxOverReservedBytes) {
         public SizeSettings(Settings settings) {
@@ -57,6 +58,7 @@ public void circuitBreak(String fieldName, long bytesNeeded) {
 
     @Override
     public void addEstimateBytesAndMaybeBreak(long bytes, String label) throws CircuitBreakingException {
+        assert assertSingleThread();
         if (bytes <= reservedBytes) {
             reservedBytes -= bytes;
             maybeReduceReservedBytes();
@@ -68,6 +70,7 @@ public void addEstimateBytesAndMaybeBreak(long bytes, String label) throws Circu
 
     @Override
     public void addWithoutBreaking(long bytes) {
+        assert assertSingleThread();
         if (bytes <= reservedBytes) {
             reservedBytes -= bytes;
             maybeReduceReservedBytes();
@@ -130,6 +133,7 @@ public void setLimitAndOverhead(long limit, double overhead) {
 
     @Override
     public void close() {
+        assert assertSingleThread();
         if (closed.compareAndSet(false, true)) {
             breaker.addWithoutBreaking(-reservedBytes);
         }
@@ -139,4 +143,34 @@ public void close() {
     public String toString() {
         return "LocalCircuitBreaker[" + reservedBytes + "/" + overReservedBytes + ":" + maxOverReservedBytes + "]";
     }
+
+    private boolean assertSingleThread() {
+        Thread activeThread = this.activeThread;
+        Thread currentThread = Thread.currentThread();
+        assert activeThread == null || activeThread == currentThread
+            : "Local breaker must be accessed by a single thread at a time: expected ["
+                + activeThread
+                + "] != actual ["
+                + currentThread
+                + "]";
+        return true;
+    }
+
+    /**
+     * Marks the beginning of a run loop for assertion purposes.
+     * Sets the current thread as the only thread allowed to access this breaker.
+     */
+    public boolean assertBeginRunLoop() {
+        activeThread = Thread.currentThread();
+        return true;
+    }
+
+    /**
+     * Marks the end of a run loop for assertion purposes.
+     * Clears the active thread to allow other threads to access this breaker.
+     */
+    public boolean assertEndRunLoop() {
+        activeThread = null;
+        return true;
+    }
 }

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/lucene/ShardContext.java

@@ -53,4 +53,10 @@ public interface ShardContext {
      * Returns something to load values from this field into a {@link Block}.
      */
     BlockLoader blockLoader(String name, boolean asUnsupportedSource, MappedFieldType.FieldExtractPreference fieldExtractPreference);
+
+    /**
+     * Returns the {@link MappedFieldType} for the given field name.
+     * By default, this delegate to {@link org.elasticsearch.index.query.SearchExecutionContext#getFieldType(String)}
+     */
+    MappedFieldType fieldType(String name);
 }