natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use OpOrder in repairIterator to ensure we don't lose memtables mid-paxos repair 

Patch by Jon Haddad; Reviewed by Blake Eggleston for CASSANDRA-19668
This commit is contained in:
Jon Haddad 2024-05-30 08:24:22 -07:00 committed by Blake Eggleston
parent ef8bc82f1b
commit 2112c4c1c0
2 changed files with 18 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.1.6
* Use OpOrder in repairIterator to ensure we don't lose memtables mid-paxos repair (Cassandra-19668)
* Refresh stale paxos commit (CASSANDRA-19617)
* Reduce info logging from automatic paxos repair (CASSANDRA-19445)
* Support legacy plain_text_auth section in credentials file removed unintentionally (CASSANDRA-19498)

30
src/java/org/apache/cassandra/service/paxos/uncommitted/PaxosUncommittedIndex.java
View File

@ -49,6 +49,7 @@ import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.Indexes;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.concurrent.OpOrder;
import static java.util.Collections.*;
import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME;
@ -135,21 +136,24 @@ public class PaxosUncommittedIndex implements Index, PaxosUncommittedTracker.Upd
{
Preconditions.checkNotNull(tableId);
View view = baseCfs.getTracker().getView();
List<Memtable> memtables = view.flushingMemtables.isEmpty()
? view.liveMemtables
: ImmutableList.<Memtable>builder().addAll(view.flushingMemtables).addAll(view.liveMemtables).build();
List<DataRange> dataRanges = ranges.stream().map(DataRange::forTokenRange).collect(Collectors.toList());
List<UnfilteredPartitionIterator> iters = new ArrayList<>(memtables.size() * ranges.size());
for (int j=0, jsize=dataRanges.size(); j<jsize; j++)
try(OpOrder.Group op = baseCfs.readOrdering.start())
{
for (int i=0, isize=memtables.size(); i<isize; i++)
iters.add(memtables.get(i).partitionIterator(memtableColumnFilter, dataRanges.get(j), SSTableReadsListener.NOOP_LISTENER));
}
View view = baseCfs.getTracker().getView();
return getPaxosUpdates(iters, tableId, false);
List<Memtable> memtables = view.flushingMemtables.isEmpty()
? view.liveMemtables
: ImmutableList.<Memtable>builder().addAll(view.flushingMemtables).addAll(view.liveMemtables).build();
List<DataRange> dataRanges = ranges.stream().map(DataRange::forTokenRange).collect(Collectors.toList());
List<UnfilteredPartitionIterator> iters = new ArrayList<>(memtables.size() * ranges.size());
for (int j = 0, jsize = dataRanges.size(); j < jsize; j++)
{
for (int i = 0, isize = memtables.size(); i < isize; i++)
iters.add(memtables.get(i).partitionIterator(memtableColumnFilter, dataRanges.get(j), SSTableReadsListener.NOOP_LISTENER));
}
return getPaxosUpdates(iters, tableId, false);
}
}
public CloseableIterator<PaxosKeyState> flushIterator(Memtable flushing)