CEP-45: Fix OpOrder.Group leak in ReadExecutionController.forCommand()

src/java/org/apache/cassandra/db/ReadExecutionController.java

@@ -158,17 +158,20 @@ static ReadExecutionController forCommand(ReadCommand command, boolean trackRepa
 
         long createdAtNanos = baseCfs.metric.topLocalReadQueryTime.isEnabled() ? clock.now() : NO_SAMPLING;
 
-        if (indexCfs == null)
-            return new ReadExecutionController(command, baseCfs.readOrdering.start(), baseCfs.metadata(), null, null, createdAtNanos, trackRepairedStatus);
-
         OpOrder.Group baseOp = null;
+        OpOrder.Group indexBaseOp = null;
         WriteContext writeContext = null;
         ReadExecutionController indexController = null;
         // OpOrder.start() shouldn't fail, but better safe than sorry.
         try
         {
             baseOp = baseCfs.readOrdering.start();
-            indexController = new ReadExecutionController(command, indexCfs.readOrdering.start(), indexCfs.metadata(), null, null, NO_SAMPLING, false);
+
+            if (indexCfs == null)
+                return new ReadExecutionController(command, baseOp, baseCfs.metadata(), null, null, createdAtNanos, trackRepairedStatus);
+
+            indexBaseOp = indexCfs.readOrdering.start();
+            indexController = new ReadExecutionController(command, indexBaseOp, indexCfs.metadata(), null, null, NO_SAMPLING, false);
             /*
              * TODO: this should perhaps not open and maintain a writeOp for the full duration, but instead only *try*
              * to delete stale entries, without blocking if there's no room
@@ -179,8 +182,6 @@ static ReadExecutionController forCommand(ReadCommand command, boolean trackRepa
         }
         catch (RuntimeException e)
         {
-            // Note that must have writeContext == null since ReadOrderGroup ctor can't fail
-            assert writeContext == null;
             try
             {
                 if (baseOp != null)
@@ -189,7 +190,21 @@ static ReadExecutionController forCommand(ReadCommand command, boolean trackRepa
             finally
             {
                 if (indexController != null)
-                    indexController.close();
+                {
+                    try
+                    {
+                        indexController.close();
+                    }
+                    finally
+                    {
+                        if (writeContext != null)
+                            writeContext.close();
+                    }
+                }
+                else if (indexBaseOp != null)
+                {
+                    indexBaseOp.close();
+                }
             }
             throw e;
         }

src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java

@@ -149,9 +149,11 @@ public void startRepair(DigestResolver<E, P> digestResolver, Consumer<PartitionI
          * pending repair session which has been committed. In addition to the digest, a set of ids for any pending but 
          * as yet uncommitted repair sessions is recorded and returned to the coordinator. This is to help reduce false 
          * positives caused by compaction lagging which can leave sstables from committed sessions in the pending state
-         * for a time.
+         * for a time. We don't track the repaired status for tables whose replication type is tracked which purely
+         * relies on the mutation-tracking log to track reconciliation of mutations.
          */
-        boolean trackRepairedStatus = DatabaseDescriptor.getRepairedDataTrackingForPartitionReadsEnabled();
+        boolean trackRepairedStatus = DatabaseDescriptor.getRepairedDataTrackingForPartitionReadsEnabled()
+                                      && !command.metadata().replicationType().isTracked();
 
         // Do a full data read to resolve the correct response (and repair node that need be)
         DataResolver<E, P> resolver = new DataResolver<>(coordinator, command, replicaPlan, this, requestTime, trackRepairedStatus);

test/unit/org/apache/cassandra/db/ReadExecutionControllerTest.java

@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.apache.cassandra.db;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.Set;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
+import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.filter.DataLimits;
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
+import org.apache.cassandra.index.Index;
+import org.apache.cassandra.index.StubIndex;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.schema.IndexMetadata;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.ReplicationType;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.Tables;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.AssertUtil;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Unit tests for {@link ReadExecutionController}
+ */
+public class ReadExecutionControllerTest
+{
+    private static final String KEYSPACE = "read_execution_controller_test_ks";
+    private static final String TABLE = "table1";
+    private static final String INDEX_TABLE = "table1_index";
+
+    private static ColumnFamilyStore cfs;
+    private static ColumnFamilyStore indexCfs;
+
+    @BeforeClass
+    public static void init() throws Exception
+    {
+        ServerTestUtils.prepareServerNoRegister();
+        MutationJournal.start();
+
+        TableMetadata table = TableMetadata.builder(KEYSPACE, TABLE)
+                                           .addPartitionKeyColumn("pk", Int32Type.instance)
+                                           .addRegularColumn("v1", Int32Type.instance)
+                                           .build();
+        TableMetadata indexTable = TableMetadata.builder(KEYSPACE, INDEX_TABLE)
+                                                .addPartitionKeyColumn("pk", Int32Type.instance)
+                                                .addRegularColumn("v1", Int32Type.instance)
+                                                .build();
+        KeyspaceMetadata ks = KeyspaceMetadata.create(KEYSPACE,
+                                                      KeyspaceParams.simple(1, ReplicationType.tracked),
+                                                      Tables.of(table, indexTable));
+        ClusterMetadataTestHelper.addOrUpdateKeyspace(ks);
+        ServerTestUtils.markCMS();
+        StorageService.instance.unsafeSetInitialized();
+
+        cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
+        indexCfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(INDEX_TABLE);
+    }
+
+    @Test
+    public void forCommandDoesNotLeakOpOrderGroupWhenRejectingTrackedTable()
+    {
+        ReadCommand command = SinglePartitionReadCommand.fullPartitionRead(cfs.metadata(), FBUtilities.nowInSeconds(), Util.dk(1));
+
+        try
+        {
+            ReadExecutionController.forCommand(command, true);
+            fail("Expected forCommand to throw because tracking the repaired status is unsupported for tracked reads");
+        }
+        catch (IllegalArgumentException e)
+        {
+            assertTrue("Unexpected exception message: " + e.getMessage(),
+                       e.getMessage().contains("Tracking repaired status is not supported for tracked reads"));
+        }
+
+        AssertUtil.assertTimeoutPreemptively(Duration.ofSeconds(5), () -> {
+            OpOrder.Barrier barrier = cfs.readOrdering.newBarrier();
+            barrier.issue();
+            barrier.await();
+        });
+    }
+
+    @Test
+    public void forCommandDoesNotLeakOpOrderGroupWhenRejectingTrackedTableWithIndex()
+    {
+        Index index = new BackedStubIndex(cfs, IndexMetadata.fromSchemaMetadata("stub_index", IndexMetadata.Kind.CUSTOM, Collections.emptyMap()));
+        Index.QueryPlan queryPlan = new SingleIndexQueryPlan(index);
+
+        ReadCommand command = SinglePartitionReadCommand.create(cfs.metadata(),
+                                                                FBUtilities.nowInSeconds(),
+                                                                ColumnFilter.all(cfs.metadata()),
+                                                                RowFilter.none(),
+                                                                DataLimits.NONE,
+                                                                Util.dk(1),
+                                                                new ClusteringIndexSliceFilter(Slices.ALL, false),
+                                                                queryPlan);
+
+        try
+        {
+            ReadExecutionController.forCommand(command, true);
+            fail("Expected forCommand to throw because tracking the repaired status is unsupported for tracked reads");
+        }
+        catch (IllegalArgumentException e)
+        {
+            assertTrue("Unexpected exception message: " + e.getMessage(),
+                       e.getMessage().contains("Tracking repaired status is not supported for tracked reads"));
+        }
+
+        // baseOp must have been released despite the exception
+        AssertUtil.assertTimeoutPreemptively(Duration.ofSeconds(5), () -> {
+            OpOrder.Barrier barrier = cfs.readOrdering.newBarrier();
+            barrier.issue();
+            barrier.await();
+        });
+        // ensure the index's OpOrder is released too
+        AssertUtil.assertTimeoutPreemptively(Duration.ofSeconds(5), () -> {
+            OpOrder.Barrier barrier = indexCfs.readOrdering.newBarrier();
+            barrier.issue();
+            barrier.await();
+        });
+    }
+
+    private static class BackedStubIndex extends StubIndex
+    {
+        BackedStubIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
+        {
+            super(baseCfs, metadata);
+        }
+
+        @Override
+        public Optional<ColumnFamilyStore> getBackingTable()
+        {
+            return Optional.of(indexCfs);
+        }
+    }
+
+    private static class SingleIndexQueryPlan implements Index.QueryPlan
+    {
+        private final Index index;
+
+        SingleIndexQueryPlan(Index index)
+        {
+            this.index = index;
+        }
+
+        @Override
+        public Set<Index> getIndexes()
+        {
+            return Collections.singleton(index);
+        }
+
+        @Override
+        public Index.Searcher searcherFor(ReadCommand command)
+        {
+            return index.searcherFor(command);
+        }
+
+        @Override
+        public RowFilter postIndexQueryFilter()
+        {
+            return RowFilter.none();
+        }
+    }
+}