You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2024/05/28 18:31:50 UTC
(accumulo) branch main updated (bd024372a1 -> 7211081310)
This is an automated email from the ASF dual-hosted git repository.
kturner pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from bd024372a1 Fix sserver group property argument in ScanServerGroupConfigurationIT (#4606)
add ea235b6c58 Invalid External Compaction Progress Fix (#4452)
add ba296913b4 Fixes 'admin fate --print/--summary' no node exception bug (#4556)
new 7211081310 Merge branch '2.1'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/accumulo/core/fate/AdminUtil.java | 62 +++++++-----
.../accumulo/core/file/BloomFilterLayer.java | 6 ++
.../apache/accumulo/core/file/FileSKVIterator.java | 11 +++
.../core/file/rfile/MultiIndexIterator.java | 6 ++
.../org/apache/accumulo/core/file/rfile/RFile.java | 60 ++++++++++--
.../iteratorsImpl/system/SequenceFileIterator.java | 6 ++
.../org/apache/accumulo/compactor/Compactor.java | 38 +++++++-
.../org/apache/accumulo/test/FateSummaryIT.java | 105 ++++++++++++++++++++-
.../compaction/ExternalCompactionProgressIT.java | 103 ++++++++++++++++++--
9 files changed, 358 insertions(+), 39 deletions(-)
(accumulo) 01/01: Merge branch '2.1'
Posted by kt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kturner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 72110813107f14949bc082a109b4e35b06ea39ad
Merge: bd024372a1 ba296913b4
Author: Keith Turner <kt...@apache.org>
AuthorDate: Tue May 28 18:31:40 2024 +0000
Merge branch '2.1'
.../org/apache/accumulo/core/fate/AdminUtil.java | 62 +++++++-----
.../accumulo/core/file/BloomFilterLayer.java | 6 ++
.../apache/accumulo/core/file/FileSKVIterator.java | 11 +++
.../core/file/rfile/MultiIndexIterator.java | 6 ++
.../org/apache/accumulo/core/file/rfile/RFile.java | 60 ++++++++++--
.../iteratorsImpl/system/SequenceFileIterator.java | 6 ++
.../org/apache/accumulo/compactor/Compactor.java | 38 +++++++-
.../org/apache/accumulo/test/FateSummaryIT.java | 105 ++++++++++++++++++++-
.../compaction/ExternalCompactionProgressIT.java | 103 ++++++++++++++++++--
9 files changed, 358 insertions(+), 39 deletions(-)
diff --cc core/src/main/java/org/apache/accumulo/core/fate/AdminUtil.java
index 7cc0a9c004,b7afea1b84..6a5873f2b3
--- a/core/src/main/java/org/apache/accumulo/core/fate/AdminUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/AdminUtil.java
@@@ -341,37 -341,57 +341,57 @@@ public class AdminUtil<T>
List<TransactionStatus> statuses = new ArrayList<>(transactions.size());
for (Long tid : transactions) {
+ try {
+ zs.reserve(tid);
- zs.reserve(tid);
-
- String txName = (String) zs.getTransactionInfo(tid, Fate.TxInfo.TX_NAME);
+ String txName = (String) zs.getTransactionInfo(tid, Fate.TxInfo.TX_NAME);
- List<String> hlocks = heldLocks.remove(tid);
+ List<String> hlocks = heldLocks.remove(tid);
- if (hlocks == null) {
- hlocks = Collections.emptyList();
- }
+ if (hlocks == null) {
+ hlocks = Collections.emptyList();
+ }
- List<String> wlocks = waitingLocks.remove(tid);
+ List<String> wlocks = waitingLocks.remove(tid);
- if (wlocks == null) {
- wlocks = Collections.emptyList();
- }
+ if (wlocks == null) {
+ wlocks = Collections.emptyList();
+ }
- String top = null;
- ReadOnlyRepo<T> repo = zs.top(tid);
- if (repo != null) {
- top = repo.getName();
- }
+ String top = null;
+ ReadOnlyRepo<T> repo = zs.top(tid);
+ if (repo != null) {
+ top = repo.getName();
+ }
- TStatus status = zs.getStatus(tid);
+ TStatus status = zs.getStatus(tid);
- long timeCreated = zs.timeCreated(tid);
+ long timeCreated = zs.timeCreated(tid);
- zs.unreserve(tid, Duration.ZERO);
- zs.unreserve(tid, 0, TimeUnit.MILLISECONDS);
++ zs.unreserve(tid, Duration.ZERO);
- if (includeByStatus(status, filterStatus) && includeByTxid(tid, filterTxid)) {
- statuses.add(new TransactionStatus(tid, status, txName, hlocks, wlocks, top, timeCreated));
+ if (includeByStatus(status, filterStatus) && includeByTxid(tid, filterTxid)) {
+ statuses
+ .add(new TransactionStatus(tid, status, txName, hlocks, wlocks, top, timeCreated));
+ }
+ } catch (Exception e) {
+ // If the cause of the Exception is a NoNodeException, it should be ignored as this
+ // indicates the transaction has completed between the time the list of transactions was
+ // acquired and the time the transaction was probed for info.
+ boolean nne = false;
+ Throwable cause = e;
+ while (cause != null) {
+ if (cause instanceof KeeperException.NoNodeException) {
+ nne = true;
+ break;
+ }
+ cause = cause.getCause();
+ }
+ if (!nne) {
+ throw e;
+ }
+ log.debug("Tried to get info on a since completed transaction - ignoring "
+ + FateTxId.formatTid(tid));
}
}
diff --cc core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java
index 65c0c0bd42,23da429535..b55f1bab69
--- a/core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java
@@@ -21,6 -21,8 +21,7 @@@ package org.apache.accumulo.core.file
import java.io.DataInputStream;
import java.io.IOException;
-import org.apache.accumulo.core.data.Key;
+ import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.file.blockfile.impl.CacheProvider;
import org.apache.accumulo.core.iteratorsImpl.system.InterruptibleIterator;
import org.apache.accumulo.core.sample.impl.SamplerConfigurationImpl;
diff --cc core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java
index c99d137b22,059eac98ca..2772e91659
--- a/core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/RFile.java
@@@ -1165,9 -1159,14 +1166,14 @@@ public class RFile
public void setCacheProvider(CacheProvider cacheProvider) {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public long estimateOverlappingEntries(KeyExtent extent) throws IOException {
+ throw new UnsupportedOperationException();
+ }
}
- public static class Reader extends HeapIterator implements FileSKVIterator {
+ public static class Reader extends HeapIterator implements RFileSKVIterator {
private final CachableBlockFile.Reader reader;
@@@ -1565,229 -1562,34 +1571,269 @@@
reader.setCacheProvider(cacheProvider);
}
+ @Override
+ public long estimateOverlappingEntries(KeyExtent extent) throws IOException {
+ long totalEntries = 0;
+ Key startKey = extent.toDataRange().getStartKey();
+ IndexEntry indexEntry;
+
+ for (LocalityGroupReader lgr : currentReaders) {
+ boolean prevEntryOverlapped = false;
+ var indexIter = startKey == null ? lgr.getIndex() : lgr.index.lookup(startKey);
+
+ while (indexIter.hasNext()) {
+ indexEntry = indexIter.next();
+ if (extent.contains(indexEntry.getKey().getRow())) {
+ totalEntries += indexEntry.getNumEntries();
+ prevEntryOverlapped = true;
+ } else if (prevEntryOverlapped) {
+ // The last index entry included in the count is the one after the last contained by the
+ // extent. This is because it is possible for the extent to overlap this index entry
+ // but there is no way to check whether it does or not. The index entry only contains
+ // info about the last key, but the extent may overlap but not with the last key.
+ totalEntries += indexEntry.getNumEntries();
+ prevEntryOverlapped = false;
+ break;
+ }
+ }
+ }
+
+ return totalEntries;
+ }
++
+ @Override
+ public void reset() {
+ clear();
+ }
+ }
+
+ public interface RFileSKVIterator extends FileSKVIterator {
+ FileSKVIterator getIndex() throws IOException;
+
+ void reset();
+ }
+
+ static abstract class FencedFileSKVIterator implements FileSKVIterator {
+
+ private final FileSKVIterator reader;
+ protected final Range fence;
+ private final Key fencedStartKey;
+ private final Supplier<Key> fencedEndKey;
+
+ public FencedFileSKVIterator(FileSKVIterator reader, Range fence) {
+ this.reader = Objects.requireNonNull(reader);
+ this.fence = Objects.requireNonNull(fence);
+ this.fencedStartKey = fence.getStartKey();
+ this.fencedEndKey = Suppliers.memoize(() -> getEndKey(fence.getEndKey()));
+ }
+
+ @Override
+ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
+ IteratorEnvironment env) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean hasTop() {
+ return reader.hasTop();
+ }
+
+ @Override
+ public void next() throws IOException {
+ reader.next();
+ }
+
+ @Override
+ public Key getTopKey() {
+ return reader.getTopKey();
+ }
+
+ @Override
+ public Value getTopValue() {
+ return reader.getTopValue();
+ }
+
+ @Override
+ public Text getFirstRow() throws IOException {
+ var row = reader.getFirstRow();
+ if (row != null && fence.beforeStartKey(new Key(row))) {
+ return fencedStartKey.getRow();
+ } else {
+ return row;
+ }
+ }
+
+ @Override
+ public Text getLastRow() throws IOException {
+ var row = reader.getLastRow();
+ if (row != null && fence.afterEndKey(new Key(row))) {
+ return fencedEndKey.get().getRow();
+ } else {
+ return row;
+ }
+ }
+
+ @Override
+ public boolean isRunningLowOnMemory() {
+ return reader.isRunningLowOnMemory();
+ }
+
+ @Override
+ public void setInterruptFlag(AtomicBoolean flag) {
+ reader.setInterruptFlag(flag);
+ }
+
+ @Override
+ public DataInputStream getMetaStore(String name) throws IOException {
+ return reader.getMetaStore(name);
+ }
+
+ @Override
+ public void closeDeepCopies() throws IOException {
+ reader.closeDeepCopies();
+ }
+
+ @Override
+ public void setCacheProvider(CacheProvider cacheProvider) {
+ reader.setCacheProvider(cacheProvider);
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ private Key getEndKey(Key key) {
+ // If they key is infinite it will be null or if inclusive we can just use it as is
+ // as it would be the correct value for getLastKey()
+ if (fence.isInfiniteStopKey() || fence.isEndKeyInclusive()) {
+ return key;
+ }
+
+ // If exclusive we need to strip the last byte to get the last key that is part of the
+ // actual range to return
+ final byte[] ba = key.getRow().getBytes();
+ Preconditions.checkArgument(ba.length > 0 && ba[ba.length - 1] == (byte) 0x00);
+ byte[] fba = new byte[ba.length - 1];
+ System.arraycopy(ba, 0, fba, 0, ba.length - 1);
+
+ return new Key(fba);
+ }
+
+ }
+
+ static class FencedIndex extends FencedFileSKVIterator {
+ private final FileSKVIterator source;
+
+ public FencedIndex(FileSKVIterator source, Range seekFence) {
+ super(source, seekFence);
+ this.source = source;
+ }
+
+ @Override
+ public boolean hasTop() {
+ // this code filters out data because the rfile index iterators do not support seek
+
+ // If startKey is set then discard everything until we reach the start
+ // of the range
+ if (fence.getStartKey() != null) {
+
+ while (source.hasTop() && fence.beforeStartKey(source.getTopKey())) {
+ try {
+ source.next();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ }
+
+ // If endKey is set then ensure that the current key is not passed the end of the range
+ return source.hasTop() && !fence.afterEndKey(source.getTopKey());
+ }
+
++ @Override
++ public long estimateOverlappingEntries(KeyExtent extent) throws IOException {
++ throw new UnsupportedOperationException();
++ }
++
+ @Override
+ public FileSKVIterator getSample(SamplerConfigurationImpl sampleConfig) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
+ throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ static class FencedReader extends FencedFileSKVIterator implements RFileSKVIterator {
+
+ private final Reader reader;
+
+ public FencedReader(Reader reader, Range seekFence) {
+ super(reader, seekFence);
+ this.reader = reader;
+ }
+
+ @Override
+ public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
+ throws IOException {
+ reader.reset();
+
+ if (fence != null) {
+ range = fence.clip(range, true);
+ if (range == null) {
+ return;
+ }
+ }
+
+ reader.seek(range, columnFamilies, inclusive);
+ }
+
+ @Override
+ public FencedReader deepCopy(IteratorEnvironment env) {
+ return new FencedReader(reader.deepCopy(env), fence);
+ }
+
+ @Override
+ public FileSKVIterator getIndex() throws IOException {
+ return new FencedIndex(reader.getIndex(), fence);
+ }
+
++ @Override
++ public long estimateOverlappingEntries(KeyExtent extent) throws IOException {
++ return reader.estimateOverlappingEntries(extent);
++ }
++
+ @Override
+ public FileSKVIterator getSample(SamplerConfigurationImpl sampleConfig) {
+ final Reader sample = reader.getSample(sampleConfig);
+ return sample != null ? new FencedReader(sample, fence) : null;
+ }
+
+ @Override
+ public void reset() {
+ reader.reset();
+ }
+ }
+
+ public static RFileSKVIterator getReader(final CachableBuilder cb, final TabletFile dataFile)
+ throws IOException {
+ final RFile.Reader reader = new RFile.Reader(Objects.requireNonNull(cb));
+ return dataFile.hasRange() ? new FencedReader(reader, dataFile.getRange()) : reader;
+ }
+
+ public static RFileSKVIterator getReader(final CachableBuilder cb, Range range)
+ throws IOException {
+ final RFile.Reader reader = new RFile.Reader(Objects.requireNonNull(cb));
+ return !range.isInfiniteStartKey() || !range.isInfiniteStopKey()
+ ? new FencedReader(reader, range) : reader;
}
}
diff --cc server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
index ebd195a22f,2b85401516..fbba00bab6
--- a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
+++ b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
@@@ -18,13 -18,14 +18,14 @@@
*/
package org.apache.accumulo.compactor;
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.TimeUnit.MINUTES;
-import static org.apache.accumulo.core.util.UtilWaitThread.sleepUninterruptibly;
+import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
import java.io.IOException;
+ import java.io.UncheckedIOException;
import java.net.UnknownHostException;
-import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@@ -61,18 -61,16 +62,20 @@@ import org.apache.accumulo.core.conf.Pr
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.fate.zookeeper.ServiceLock;
-import org.apache.accumulo.core.fate.zookeeper.ServiceLock.LockLossReason;
-import org.apache.accumulo.core.fate.zookeeper.ServiceLock.LockWatcher;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
+ import org.apache.accumulo.core.file.FileOperations;
+ import org.apache.accumulo.core.file.FileSKVIterator;
import org.apache.accumulo.core.iteratorsImpl.system.SystemIteratorUtil;
+import org.apache.accumulo.core.lock.ServiceLock;
+import org.apache.accumulo.core.lock.ServiceLock.LockLossReason;
+import org.apache.accumulo.core.lock.ServiceLock.LockWatcher;
+import org.apache.accumulo.core.lock.ServiceLockData;
+import org.apache.accumulo.core.lock.ServiceLockData.ServiceDescriptor;
+import org.apache.accumulo.core.lock.ServiceLockData.ServiceDescriptors;
+import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
+import org.apache.accumulo.core.metadata.ReferencedTabletFile;
import org.apache.accumulo.core.metadata.StoredTabletFile;
-import org.apache.accumulo.core.metadata.TabletFile;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
@@@ -105,7 -108,7 +109,8 @@@ import org.apache.accumulo.server.fs.Vo
import org.apache.accumulo.server.rpc.ServerAddress;
import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.rpc.ThriftProcessorTypes;
+import org.apache.accumulo.server.zookeeper.TransactionWatcher;
+ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
@@@ -537,6 -555,27 +548,27 @@@ public class Compactor extends Abstract
};
}
+ /**
+ * @param extent the extent
+ * @param file the file to read from
+ * @param tableConf the table configuration
+ * @param cryptoService the crypto service
+ * @return an estimate of the number of key/value entries in the file that overlap the extent
+ */
+ private long estimateOverlappingEntries(KeyExtent extent, StoredTabletFile file,
+ AccumuloConfiguration tableConf, CryptoService cryptoService) {
+ FileOperations fileFactory = FileOperations.getInstance();
+ FileSystem fs = getContext().getVolumeManager().getFileSystemByPath(file.getPath());
+
+ try (FileSKVIterator reader =
- fileFactory.newReaderBuilder().forFile(file.getPathStr(), fs, fs.getConf(), cryptoService)
++ fileFactory.newReaderBuilder().forFile(file, fs, fs.getConf(), cryptoService)
+ .withTableConfiguration(tableConf).dropCachesBehind().build()) {
+ return reader.estimateOverlappingEntries(extent);
+ } catch (IOException ioe) {
+ throw new UncheckedIOException(ioe);
+ }
+ }
+
/**
* Returns the number of seconds to wait in between progress checks based on input file sizes
*