ReadCommandController should close fast to avoid deadlock when buildi…

src/java/org/apache/cassandra/index/SecondaryIndexManager.java

@@ -33,6 +33,8 @@
 import com.google.common.collect.*;
 import com.google.common.primitives.Longs;
 import com.google.common.util.concurrent.FutureCallback;
+
+import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
@@ -918,14 +920,21 @@ public void indexPartition(DecoratedKey key, Set<Index> indexes, int pageSize, R
             SinglePartitionPager pager = new SinglePartitionPager(cmd, null, ProtocolVersion.CURRENT);
             while (!pager.isExhausted())
             {
+                UnfilteredRowIterator partition;
                 try (ReadExecutionController controller = cmd.executionController();
-                     WriteContext ctx = keyspace.getWriteHandler().createContextForIndexing();
                      UnfilteredPartitionIterator page = pager.fetchPageUnfiltered(baseCfs.metadata(), pageSize, controller))
                 {
                     if (!page.hasNext())
                         break;
 
-                    try (UnfilteredRowIterator partition = page.next())
+                    try (UnfilteredRowIterator onePartition = page.next())
+                    {
+                        partition = ImmutableBTreePartition.create(onePartition).unfilteredIterator();
+                    }
+                }
+
+                try (WriteContext ctx = keyspace.getWriteHandler().createContextForIndexing())
+                {
                     {
                         Set<Index.Indexer> indexers = indexes.stream()
                                                              .map(index -> index.indexerFor(key,

test/distributed/org/apache/cassandra/distributed/test/SecondaryIndexTest.java

@@ -23,12 +23,20 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Random;
 import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import com.google.common.collect.Sets;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
 import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -41,11 +49,13 @@
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.utils.TimeUUID;
 
+import static org.apache.cassandra.distributed.impl.IsolatedExecutor.waitOn;
+
 public class SecondaryIndexTest extends TestBaseImpl
 {
     private static final int NUM_NODES = 3;
     private static final int REPLICATION_FACTOR = 1;
-    private static final String CREATE_TABLE = "CREATE TABLE %s(k int, v int, PRIMARY KEY (k))";
+    private static final String CREATE_TABLE = "CREATE TABLE %s(k int, v text, PRIMARY KEY (k))";
     private static final String CREATE_INDEX = "CREATE INDEX v_index_%d ON %s(v)";
 
     private static final AtomicInteger seq = new AtomicInteger();
@@ -122,4 +132,53 @@ public void test_only_coordinator_chooses_index_for_query()
                                    });
         }
     }
+
+    @Test
+    public void test_secondary_rebuild_with_small_memtable_memory()
+    {
+        // populate data
+        for (int i = 0 ; i < 100 ; ++i)
+            cluster.coordinator(1).execute(String.format("INSERT INTO %s (k, v) VALUES (?, ?)", tableName), ConsistencyLevel.ALL, i, generateRandomString(50000));
+
+        cluster.forEach(i -> i.flush(KEYSPACE));
+
+        // restart node 1 with small memtable allocation so that index rebuild will cause memtable flush which will need
+        // to reclaim the memory. see CASSANDRA-19564
+        waitOn(cluster.get(1).shutdown());
+        cluster.get(1).config().set("memtable_heap_space", "1MiB");
+        cluster.get(1).startup();
+        String tableNameWithoutKeyspaceName = tableName.split("\\.")[1];
+        String indexName = String.format("v_index_%d", seq.get());
+        Runnable task = cluster.get(1).runsOnInstance(
+        () -> {
+            ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(tableNameWithoutKeyspaceName);
+            cfs.indexManager.rebuildIndexesBlocking(Sets.newHashSet(Arrays.asList(indexName)));
+        }
+        );
+        ExecutorService es = Executors.newFixedThreadPool(1);
+        Future future = es.submit(task);
+        try
+        {
+            future.get(30, TimeUnit.SECONDS);
+        }
+        catch (Exception e)
+        {
+            e.printStackTrace();
+            Assert.fail("Rebuild should finish within 30 seconds without issue.");
+        }
+    }
+
+    private String generateRandomString(int length) {
+        String characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
+        Random random = new Random();
+        StringBuilder sb = new StringBuilder(length);
+
+        for (int i = 0; i < length; i++) {
+            int randomIndex = random.nextInt(characters.length());
+            char randomChar = characters.charAt(randomIndex);
+            sb.append(randomChar);
+        }
+
+        return sb.toString();
+    }
 }