This is an automated email from the ASF dual-hosted git repository.
frankgh pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new e71bdecb62 CEP-45: Fix OpOrder.Group leak in
ReadExecutionController.forCommand()
e71bdecb62 is described below
commit e71bdecb62b5f774304816ad79a3641223ec3d4b
Author: Francisco Guerrero <[email protected]>
AuthorDate: Thu Aug 20 16:43:54 2026 -0700
CEP-45: Fix OpOrder.Group leak in ReadExecutionController.forCommand()
patch by Francisco Guerrero; reviewed by Mick Semb Wever for CASSANDRA-21593
---
.../cassandra/db/ReadExecutionController.java | 29 ++-
.../service/reads/repair/AbstractReadRepair.java | 6 +-
.../cassandra/db/ReadExecutionControllerTest.java | 194 +++++++++++++++++++++
3 files changed, 220 insertions(+), 9 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/ReadExecutionController.java
b/src/java/org/apache/cassandra/db/ReadExecutionController.java
index 513199fdd5..5d9b636dcb 100644
--- a/src/java/org/apache/cassandra/db/ReadExecutionController.java
+++ b/src/java/org/apache/cassandra/db/ReadExecutionController.java
@@ -158,17 +158,20 @@ public class ReadExecutionController implements
AutoCloseable
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 @@ public class ReadExecutionController implements
AutoCloseable
}
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 @@ public class ReadExecutionController implements
AutoCloseable
finally
{
if (indexController != null)
- indexController.close();
+ {
+ try
+ {
+ indexController.close();
+ }
+ finally
+ {
+ if (writeContext != null)
+ writeContext.close();
+ }
+ }
+ else if (indexBaseOp != null)
+ {
+ indexBaseOp.close();
+ }
}
throw e;
}
diff --git
a/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java
b/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java
index a84259ee9e..2e578c595e 100644
--- a/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java
+++ b/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java
@@ -149,9 +149,11 @@ public abstract class AbstractReadRepair<E extends
Endpoints<E>, P extends Repli
* 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);
diff --git a/test/unit/org/apache/cassandra/db/ReadExecutionControllerTest.java
b/test/unit/org/apache/cassandra/db/ReadExecutionControllerTest.java
new file mode 100644
index 0000000000..5e836c58ac
--- /dev/null
+++ b/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();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
