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

Fix closing iterator in SecondaryIndexBuilder 

Patch by Jacek Lewandowski; reviewed by Andres de la Peña, Piotr Kolaczkowski for CASSANDRA-18361
This commit is contained in:
Jacek Lewandowski 2023-09-28 14:23:42 +02:00
parent 9ccec3dc8c
commit 016d91a7d7
4 changed files with 55 additions and 20 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.0.12
* Fix closing iterator in SecondaryIndexBuilder (CASSANDRA-18361)
* Update hdrhistogram to 2.1.12 (CASSANDRA-18893)
* Improve performance of compactions when table does not have an index (CASSANDRA-18773)
* JMH improvements - faster build and async profiler (CASSANDRA-18871)

22
src/java/org/apache/cassandra/index/SecondaryIndexManager.java
View File

@ -41,6 +41,7 @@ import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import org.apache.cassandra.utils.Throwables;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
@ -519,17 +520,34 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
final SettableFuture build = SettableFuture.create();
Futures.addCallback(CompactionManager.instance.submitIndexBuild(builder), new FutureCallback()
{
@Override
public void onFailure(Throwable t)
private void doOnFailure(Throwable t)
{
logAndMarkIndexesFailed(groupedIndexes, t, false);
unbuiltIndexes.addAll(groupedIndexes);
build.setException(t);
}
@Override
public void onFailure(Throwable t)
{
if (builder instanceof AutoCloseable)
t = Throwables.close(t, Arrays.asList((AutoCloseable) builder));
doOnFailure(t);
}
@Override
public void onSuccess(Object o)
{
if (builder instanceof AutoCloseable)
{
Throwable t = Throwables.close(null, Arrays.asList((AutoCloseable) builder));
if (t != null)
{
doOnFailure(t);
return;
}
}
groupedIndexes.forEach(i -> markIndexBuilt(i, isFullRebuild));
logger.info("Index build of {} completed", getIndexNames(groupedIndexes));
builtIndexes.addAll(groupedIndexes);

36
src/java/org/apache/cassandra/index/internal/CassandraIndex.java
View File

@ -23,19 +23,18 @@ package org.apache.cassandra.index.internal;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.db.*;
@ -57,7 +56,10 @@ import org.apache.cassandra.index.transactions.IndexTransaction;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Refs;
@ -703,11 +705,25 @@ public abstract class CassandraIndex implements Index
metadata.name,
getSSTableNames(sstables));
SecondaryIndexBuilder builder = new CollatedViewIndexBuilder(baseCfs,
Collections.singleton(this),
new ReducingKeyIterator(sstables),
ImmutableSet.copyOf(sstables));
Future<?> future = CompactionManager.instance.submitIndexBuild(builder);
CollatedViewIndexBuilder builder = new CollatedViewIndexBuilder(baseCfs,
Collections.singleton(this),
new ReducingKeyIterator(sstables),
ImmutableSet.copyOf(sstables));
ListenableFuture<?> future = CompactionManager.instance.submitIndexBuild(builder);
Futures.addCallback(future, new FutureCallback<Object>()
{
@Override
public void onSuccess(Object o)
{
builder.close();
}
@Override
public void onFailure(Throwable throwable)
{
builder.close();
}
}, MoreExecutors.directExecutor());
FBUtilities.waitOnFuture(future);
indexCfs.forceBlockingFlush();
}

16
src/java/org/apache/cassandra/index/internal/CollatedViewIndexBuilder.java
View File

@ -35,7 +35,7 @@ import org.apache.cassandra.utils.UUIDGen;
/**
* Manages building an entire index from column family data. Runs on to compaction manager.
*/
public class CollatedViewIndexBuilder extends SecondaryIndexBuilder
public class CollatedViewIndexBuilder extends SecondaryIndexBuilder implements AutoCloseable
{
private final ColumnFamilyStore cfs;
private final Set<Index> indexers;
@ -64,8 +64,7 @@ public class CollatedViewIndexBuilder extends SecondaryIndexBuilder
public void build()
{
try
{
int pageSize = cfs.indexManager.calculateIndexingPageSize();
while (iter.hasNext())
{
@ -73,11 +72,12 @@ public class CollatedViewIndexBuilder extends SecondaryIndexBuilder
throw new CompactionInterruptedException(getCompactionInfo());
DecoratedKey key = iter.next();
cfs.indexManager.indexPartition(key, indexers, pageSize);
}
}
finally
{
iter.close();
}
}
@Override
public void close()
{
iter.close();
}
}