/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch 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.elasticsearch.indices.state;

import io.crate.execution.ddl.tables.TransportCloseTable;
import io.crate.integrationtests.SQLIntegrationTestCase;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.index.IndexSettings;
import org.junit.Test;

import java.util.List;

import static java.util.Collections.emptySet;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;

public class CloseIndexIT extends SQLIntegrationTestCase {

    @Override
    public Settings indexSettings() {
        return Settings.builder().put(super.indexSettings())
            .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(),
                 new ByteSizeValue(randomIntBetween(1, 4096), ByteSizeUnit.KB)).build();
    }

    /**
     * Test for https://github.com/elastic/elasticsearch/issues/47276 which checks that the persisted metadata on a data node does not
     * become inconsistent when using replicated closed indices.
     */
    @Test
    public void testRelocatedClosedIndexIssue() throws Exception {
        final List<String> dataNodes = internalCluster().startDataOnlyNodes(2);
        // allocate shard to first data node
        execute("create table doc.test(x int) clustered into 1 shards with (number_of_replicas=0, \"routing.allocation.include._name\" = ?)", new Object[] {dataNodes.get(0)});
        var numDocs = randomIntBetween(0, 50);
        for(var i = 0; i < numDocs; i++) {
            execute("insert into doc.test values(?)", new Object[]{i});
        }

        execute("alter table doc.test close");
        // Closed tables cannot not be altered, therefore use the api
        client().admin().indices().prepareUpdateSettings("test").setSettings(Settings.builder().put("index.routing.allocation.include._name", dataNodes.get(1))).get();
        ensureGreen("test");
        internalCluster().fullRestart();
        assertIndexIsClosed("test");
        ensureGreen("test");
    }

    static void assertIndexIsClosed(final String... indices) {
        final ClusterState clusterState = client().admin().cluster().prepareState().get().getState();
        for (String index : indices) {
            final IndexMetadata indexMetadata = clusterState.metadata().indices().get(index);
            assertThat(indexMetadata.getState(), is(IndexMetadata.State.CLOSE));
            final Settings indexSettings = indexMetadata.getSettings();
            assertThat(indexSettings.hasValue(IndexMetadata.VERIFIED_BEFORE_CLOSE_SETTING.getKey()), is(true));
            assertThat(indexSettings.getAsBoolean(IndexMetadata.VERIFIED_BEFORE_CLOSE_SETTING.getKey(), false), is(true));
            assertThat(clusterState.routingTable().index(index), notNullValue());
            assertThat(clusterState.blocks().hasIndexBlock(index, IndexMetadata.INDEX_CLOSED_BLOCK), is(true));
            assertThat("Index " + index + " must have only 1 block with [id=" + TransportCloseTable.INDEX_CLOSED_BLOCK_ID + "]",
                       clusterState.blocks().indices().getOrDefault(index, emptySet()).stream()
                           .filter(clusterBlock -> clusterBlock.id() == TransportCloseTable.INDEX_CLOSED_BLOCK_ID).count(), equalTo(1L));
        }
    }
}

@@ -1043,16 +1043,74 @@
 tadata;%0A
+import org.elasticsearch.common.collect.ImmutableOpenMap;%0A
 import o
@@ -1293,32 +1293,62 @@
 .IndexSettings;%0A
+import org.hamcrest.Matchers;%0A
 import org.junit
@@ -2556,16 +2556,62 @@
 0, 50);%0A
+        var bulkArgs = new Object%5BnumDocs%5D%5B%5D;%0A
         
@@ -2641,32 +2641,84 @@
 ocs; i++) %7B%0A    
+        bulkArgs%5Bi%5D = new Object%5B%5D %7B i %7D;%0A        %7D%0A
         execute(
@@ -2755,35 +2755,18 @@
 )%22, 
-new Object%5B%5D%7Bi%7D);%0A        %7D
+bulkArgs);
 %0A%0A  
@@ -3130,56 +3130,56 @@
     
-assertIndexIsClosed(%22test%22);%0A        ensureGreen
+ensureGreen(%22test%22);%0A        assertIndexIsClosed
 (%22te
@@ -3266,26 +3266,11 @@
     
-final ClusterState
+var
  clu
@@ -3337,24 +3337,192 @@
 getState();%0A
+        var availableIndices = clusterState.metadata().indices();%0A        assertThat(availableIndices.keys().toArray(String.class), Matchers.arrayContaining(indices));%0A
         for 
@@ -3600,41 +3600,24 @@
 a = 
-clusterState.metadata().i
+availableI
 ndices
-()
 .get

Fix flaky testRelocatedClosedIndexIssue