You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2025/04/18 05:01:17 UTC
(accumulo) branch main updated (393faab5a9 -> 17646f77eb)
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 393faab5a9 Modified TabletMetadata to return HOSTED for all system tables (#5486)
add 3af3fde623 Fix batch scan audit logging (#5483)
new 17646f77eb 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:
.../server/client/ClientServiceHandler.java | 4 +-
.../server/security/AuditedSecurityOperation.java | 68 +++----
.../server/security/SecurityOperation.java | 97 ++++------
.../server/security/handler/ZKAuthorizor.java | 3 -
.../accumulo/manager/FateServiceHandler.java | 47 +++--
.../java/org/apache/accumulo/manager/Manager.java | 6 +-
.../manager/ManagerClientServiceHandler.java | 107 ++++++-----
.../coordinator/CompactionCoordinator.java | 12 +-
.../manager/tableOps/create/SetupPermissions.java | 3 +-
.../create/SetupNamespacePermissions.java | 3 +-
.../tableImport/ImportSetupPermissions.java | 3 +-
.../compaction/CompactionCoordinatorTest.java | 201 ++++++++-------------
.../accumulo/tserver/TabletClientHandler.java | 74 ++++----
.../accumulo/tserver/ThriftScanClientHandler.java | 25 ++-
.../accumulo/tserver/TservConstraintEnv.java | 12 +-
.../accumulo/tserver/TservConstraintEnvTest.java | 17 +-
16 files changed, 298 insertions(+), 384 deletions(-)
(accumulo) 01/01: Merge branch '2.1'
Posted by ct...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 17646f77ebb662962cd39d977f5fb57a7acc4c54
Merge: 393faab5a9 3af3fde623
Author: Christopher Tubbs <ct...@apache.org>
AuthorDate: Fri Apr 18 00:37:46 2025 -0400
Merge branch '2.1'
This includes related changes to remove redundant parameters from the
CompactionCoordinator constructor and corresponding changes to
CompactionCoordinatorTest to ensure it passes (and to clean up EasyMock
usage)
.../server/client/ClientServiceHandler.java | 4 +-
.../server/security/AuditedSecurityOperation.java | 68 +++----
.../server/security/SecurityOperation.java | 97 ++++------
.../server/security/handler/ZKAuthorizor.java | 3 -
.../accumulo/manager/FateServiceHandler.java | 47 +++--
.../java/org/apache/accumulo/manager/Manager.java | 6 +-
.../manager/ManagerClientServiceHandler.java | 107 ++++++-----
.../coordinator/CompactionCoordinator.java | 12 +-
.../manager/tableOps/create/SetupPermissions.java | 3 +-
.../create/SetupNamespacePermissions.java | 3 +-
.../tableImport/ImportSetupPermissions.java | 3 +-
.../compaction/CompactionCoordinatorTest.java | 201 ++++++++-------------
.../accumulo/tserver/TabletClientHandler.java | 74 ++++----
.../accumulo/tserver/ThriftScanClientHandler.java | 25 ++-
.../accumulo/tserver/TservConstraintEnv.java | 12 +-
.../accumulo/tserver/TservConstraintEnvTest.java | 17 +-
16 files changed, 298 insertions(+), 384 deletions(-)
diff --cc server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java
index 9bd4292e50,2edf8a55c2..b6b2f47d67
--- a/server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/client/ClientServiceHandler.java
@@@ -62,19 -65,24 +62,19 @@@ import org.apache.accumulo.server.Serve
import org.apache.accumulo.server.conf.store.NamespacePropKey;
import org.apache.accumulo.server.conf.store.SystemPropKey;
import org.apache.accumulo.server.conf.store.TablePropKey;
- import org.apache.accumulo.server.security.SecurityOperation;
+ import org.apache.accumulo.server.security.AuditedSecurityOperation;
-import org.apache.accumulo.server.util.ServerBulkImportStatus;
import org.apache.accumulo.server.util.TableDiskUsage;
-import org.apache.accumulo.server.zookeeper.TransactionWatcher;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ClientServiceHandler implements ClientService.Iface {
private static final Logger log = LoggerFactory.getLogger(ClientServiceHandler.class);
- protected final TransactionWatcher transactionWatcher;
protected final ServerContext context;
- protected final SecurityOperation security;
+ protected final AuditedSecurityOperation security;
- private final ServerBulkImportStatus bulkImportStatus = new ServerBulkImportStatus();
- public ClientServiceHandler(ServerContext context, TransactionWatcher transactionWatcher) {
+ public ClientServiceHandler(ServerContext context) {
this.context = context;
- this.transactionWatcher = transactionWatcher;
this.security = context.getSecurityOperation();
}
diff --cc server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java
index 0d10b2025f,f1068ec6ff..9739bd9fb6
--- a/server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/AuditedSecurityOperation.java
@@@ -36,10 -40,9 +40,9 @@@ import org.apache.accumulo.core.data.Ta
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.dataImpl.thrift.IterInfo;
import org.apache.accumulo.core.dataImpl.thrift.TColumn;
- import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
import org.apache.accumulo.core.dataImpl.thrift.TRange;
-import org.apache.accumulo.core.manager.thrift.FateOperation;
-import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.manager.thrift.TFateOperation;
+import org.apache.accumulo.core.metadata.AccumuloTable;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.SystemPermission;
import org.apache.accumulo.core.security.TablePermission;
diff --cc server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java
index c8529953ab,45e665c882..1d77908754
--- a/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java
@@@ -38,12 -37,10 +37,8 @@@ import org.apache.accumulo.core.clientI
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.TableId;
- import org.apache.accumulo.core.dataImpl.thrift.IterInfo;
- import org.apache.accumulo.core.dataImpl.thrift.TColumn;
- import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
- import org.apache.accumulo.core.dataImpl.thrift.TRange;
-import org.apache.accumulo.core.fate.zookeeper.ZooCache;
-import org.apache.accumulo.core.manager.thrift.FateOperation;
-import org.apache.accumulo.core.metadata.MetadataTable;
-import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.manager.thrift.TFateOperation;
+import org.apache.accumulo.core.metadata.AccumuloTable;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.NamespacePermission;
import org.apache.accumulo.core.security.SystemPermission;
@@@ -521,7 -511,7 +503,7 @@@ public class SecurityOperation
|| hasTablePermission(c, tableId, namespaceId, TablePermission.DROP_TABLE, false);
}
- public boolean canChangeTableState(TCredentials c, TableId tableId, TFateOperation op,
- protected boolean canOnlineOfflineTable(TCredentials c, TableId tableId, FateOperation op,
++ protected boolean canChangeTableState(TCredentials c, TableId tableId, TFateOperation op,
NamespaceId namespaceId) throws ThriftSecurityException {
authenticate(c);
return hasSystemPermissionWithNamespaceId(c, SystemPermission.SYSTEM, namespaceId, false)
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java
index 89428fa026,52320225ba..db35b82558
--- a/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java
@@@ -107,6 -91,7 +107,7 @@@ import org.apache.accumulo.manager.tabl
import org.apache.accumulo.manager.tableOps.tableExport.ExportTable;
import org.apache.accumulo.manager.tableOps.tableImport.ImportTable;
import org.apache.accumulo.server.client.ClientServiceHandler;
-import org.apache.accumulo.server.manager.state.MergeInfo;
++import org.apache.accumulo.server.security.AuditedSecurityOperation;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
@@@ -119,10 -102,10 +120,12 @@@ import org.slf4j.Logger
class FateServiceHandler implements FateService.Iface {
private final Manager manager;
++ private final AuditedSecurityOperation security;
protected static final Logger log = Manager.log;
public FateServiceHandler(Manager manager) {
this.manager = manager;
++ this.security = manager.getContext().getSecurityOperation();
}
@Override
@@@ -150,7 -128,7 +153,7 @@@
validateArgumentCount(arguments, tableOp, 1);
String namespace = validateName(arguments.get(0), tableOp, NEW_NAMESPACE_NAME);
-- if (!manager.security.canCreateNamespace(c)) {
++ if (!security.canCreateNamespace(c)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -169,7 -147,7 +172,7 @@@
NamespaceId namespaceId =
ClientServiceHandler.checkNamespaceId(manager.getContext(), oldName, tableOp);
-- if (!manager.security.canRenameNamespace(c, namespaceId)) {
++ if (!security.canRenameNamespace(c, namespaceId)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -187,7 -165,7 +190,7 @@@
NamespaceId namespaceId =
ClientServiceHandler.checkNamespaceId(manager.getContext(), namespace, tableOp);
-- if (!manager.security.canDeleteNamespace(c, namespaceId)) {
++ if (!security.canDeleteNamespace(c, namespaceId)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -237,7 -213,7 +240,7 @@@
TableOperationExceptionType.NAMESPACE_NOTFOUND, "");
}
-- if (!manager.security.canCreateTable(c, tableName, namespaceId)) {
++ if (!security.canCreateTable(c, tableName, namespaceId)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -278,8 -252,8 +281,7 @@@
final boolean canRename;
try {
-- canRename =
-- manager.security.canRenameTable(c, tableId, oldTableName, newTableName, namespaceId);
++ canRename = security.canRenameTable(c, tableId, oldTableName, newTableName, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, oldTableName, TableOperation.RENAME);
throw e;
@@@ -335,7 -309,7 +337,7 @@@
final boolean canCloneTable;
try {
canCloneTable =
-- manager.security.canCloneTable(c, srcTableId, tableName, namespaceId, srcNamespaceId);
++ security.canCloneTable(c, srcTableId, tableName, namespaceId, srcNamespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, srcTableId, null, TableOperation.CLONE);
throw e;
@@@ -391,7 -365,7 +393,7 @@@
final boolean canDeleteTable;
try {
-- canDeleteTable = manager.security.canDeleteTable(c, tableId, namespaceId);
++ canDeleteTable = security.canDeleteTable(c, tableId, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, tableName, TableOperation.DELETE);
throw e;
@@@ -416,8 -390,8 +418,7 @@@
final boolean canOnlineOfflineTable;
try {
-- canOnlineOfflineTable =
- manager.security.canChangeTableState(c, tableId, top, namespaceId);
- manager.security.canOnlineOfflineTable(c, tableId, op, namespaceId);
++ canOnlineOfflineTable = security.canChangeTableState(c, tableId, top, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, null, TableOperation.ONLINE);
throw e;
@@@ -446,8 -420,8 +447,7 @@@
final boolean canOnlineOfflineTable;
try {
-- canOnlineOfflineTable =
- manager.security.canChangeTableState(c, tableId, top, namespaceId);
- manager.security.canOnlineOfflineTable(c, tableId, op, namespaceId);
++ canOnlineOfflineTable = security.canChangeTableState(c, tableId, top, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, null, TableOperation.OFFLINE);
throw e;
@@@ -479,7 -453,7 +479,7 @@@
final boolean canMerge;
try {
-- canMerge = manager.security.canMerge(c, tableId, namespaceId);
++ canMerge = security.canMerge(c, tableId, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, tableName, TableOperation.MERGE);
throw e;
@@@ -516,7 -490,7 +516,7 @@@
final boolean canDeleteRange;
try {
canDeleteRange =
-- manager.security.canDeleteRange(c, tableId, tableName, startRow, endRow, namespaceId);
++ security.canDeleteRange(c, tableId, tableName, startRow, endRow, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, tableName, TableOperation.DELETE_RANGE);
throw e;
@@@ -543,7 -552,7 +543,7 @@@
final boolean canCompact;
try {
-- canCompact = manager.security.canCompact(c, tableId, namespaceId);
++ canCompact = security.canCompact(c, tableId, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, null, TableOperation.COMPACT);
throw e;
@@@ -567,7 -576,7 +567,7 @@@
final boolean canCancelCompact;
try {
-- canCancelCompact = manager.security.canCompact(c, tableId, namespaceId);
++ canCancelCompact = security.canCompact(c, tableId, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, null, TableOperation.COMPACT_CANCEL);
throw e;
@@@ -608,7 -617,7 +608,7 @@@
final boolean canImport;
try {
-- canImport = manager.security.canImport(c, tableName, exportDirs, namespaceId);
++ canImport = security.canImport(c, tableName, exportDirs, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, null, tableName, TableOperation.IMPORT);
throw e;
@@@ -638,7 -647,7 +638,7 @@@
final boolean canExport;
try {
-- canExport = manager.security.canExport(c, tableId, tableName, exportDir, namespaceId);
++ canExport = security.canExport(c, tableId, tableName, exportDir, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, tableName, TableOperation.EXPORT);
throw e;
@@@ -668,8 -677,8 +668,7 @@@
String tableName;
try {
tableName = manager.getContext().getTableName(tableId);
-- canBulkImport =
-- manager.security.canBulkImport(c, tableId, tableName, dir, null, namespaceId);
++ canBulkImport = security.canBulkImport(c, tableId, tableName, dir, null, namespaceId);
} catch (ThriftSecurityException e) {
throwIfTableMissingSecurityException(e, tableId, "", TableOperation.BULK_IMPORT);
throw e;
@@@ -686,126 -695,9 +685,126 @@@
manager.updateBulkImportStatus(dir, BulkImportState.INITIAL);
goalMessage += "Bulk import (v2) " + dir + " to " + tableName + "(" + tableId + ")";
- manager.fate().seedTransaction(op.toString(), opid,
+ manager.fate(type).seedTransaction(op, fateId,
new TraceRepo<>(new PrepBulkImport(tableId, dir, setTime)), autoCleanup, goalMessage);
break;
+ }
+ case TABLE_TABLET_AVAILABILITY: {
+ TableOperation tableOp = TableOperation.SET_TABLET_AVAILABILITY;
+ validateArgumentCount(arguments, tableOp, 3);
+ String tableName = validateName(arguments.get(0), tableOp, NOT_METADATA_TABLE);
+ TableId tableId = null;
+ try {
+ tableId = manager.getContext().getTableId(tableName);
+ } catch (TableNotFoundException e) {
+ throw new ThriftTableOperationException(null, tableName,
+ TableOperation.SET_TABLET_AVAILABILITY, TableOperationExceptionType.NOTFOUND,
+ "Table no longer exists");
+ }
+ final NamespaceId namespaceId = getNamespaceIdFromTableId(tableOp, tableId);
+
+ final boolean canSetAvailability;
+ try {
- canSetAvailability = manager.security.canAlterTable(c, tableId, namespaceId);
++ canSetAvailability = security.canAlterTable(c, tableId, namespaceId);
+ } catch (ThriftSecurityException e) {
+ throwIfTableMissingSecurityException(e, tableId, tableName,
+ TableOperation.SET_TABLET_AVAILABILITY);
+ throw e;
+ }
+ if (!canSetAvailability) {
+ throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
+ }
+
+ TRange tRange = new TRange();
+ try {
+ new TDeserializer().deserialize(tRange, ByteBufferUtil.toBytes(arguments.get(1)));
+ } catch (TException e) {
+ throw new ThriftTableOperationException(tableId.canonical(), tableName,
+ TableOperation.SET_TABLET_AVAILABILITY, TableOperationExceptionType.BAD_RANGE,
+ e.getMessage());
+ }
+ TabletAvailability tabletAvailability =
+ TabletAvailability.valueOf(ByteBufferUtil.toString(arguments.get(2)));
+
+ goalMessage += "Set availability for table: " + tableName + "(" + tableId + ") range: "
+ + tRange + " to: " + tabletAvailability.name();
+ manager.fate(type).seedTransaction(op, fateId,
+ new TraceRepo<>(new LockTable(tableId, namespaceId, tRange, tabletAvailability)),
+ autoCleanup, goalMessage);
+ break;
+ }
+ case TABLE_SPLIT: {
+ TableOperation tableOp = TableOperation.SPLIT;
+
+ int SPLIT_OFFSET = 3; // offset where split data begins in arguments list
+ if (arguments.size() < (SPLIT_OFFSET + 1)) {
+ throw new ThriftTableOperationException(null, null, tableOp,
+ TableOperationExceptionType.OTHER,
+ "Expected at least " + (SPLIT_OFFSET + 1) + " arguments, saw :" + arguments.size());
+ }
+
+ var tableId = validateTableIdArgument(arguments.get(0), tableOp, NOT_ROOT_TABLE_ID);
+ NamespaceId namespaceId = getNamespaceIdFromTableId(tableOp, tableId);
+
+ boolean canSplit;
+
+ try {
- canSplit = manager.security.canSplitTablet(c, tableId, namespaceId);
++ canSplit = security.canSplitTablet(c, tableId, namespaceId);
+ } catch (ThriftSecurityException e) {
+ throwIfTableMissingSecurityException(e, tableId, null, TableOperation.SPLIT);
+ throw e;
+ }
+
+ if (!canSplit) {
+ throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
+ }
+
+ var endRow = ByteBufferUtil.toText(arguments.get(1));
+ var prevEndRow = ByteBufferUtil.toText(arguments.get(2));
+
+ endRow = endRow.getLength() == 0 ? null : endRow;
+ prevEndRow = prevEndRow.getLength() == 0 ? null : prevEndRow;
+
+ SortedMap<Text,
+ TabletMergeability> splits = arguments.subList(SPLIT_OFFSET, arguments.size()).stream()
+ .map(TabletMergeabilityUtil::decode).collect(
+ Collectors.toMap(Pair::getFirst, Pair::getSecond, (a, b) -> a, TreeMap::new));
+
+ KeyExtent extent = new KeyExtent(tableId, endRow, prevEndRow);
+
+ Predicate<Text> outOfBoundsTest =
+ split -> !extent.contains(split) || split.equals(extent.endRow());
+
+ if (splits.keySet().stream().anyMatch(outOfBoundsTest)) {
+ splits.keySet().stream().filter(outOfBoundsTest).forEach(split -> log
+ .warn("split for {} is out of bounds : {}", extent, TextUtil.truncate(split)));
+
+ throw new ThriftTableOperationException(tableId.canonical(), null, tableOp,
+ TableOperationExceptionType.OTHER,
+ "Split is outside bounds of tablet or equal to the tablets endrow, see warning in logs for more information.");
+ }
+
+ var maxSplitSize = manager.getContext().getTableConfiguration(tableId)
+ .getAsBytes(Property.TABLE_MAX_END_ROW_SIZE);
+
+ Predicate<Text> oversizedTest = split -> split.getLength() > maxSplitSize;
+
+ if (splits.keySet().stream().anyMatch(oversizedTest)) {
+ splits.keySet().stream().filter(oversizedTest)
+ .forEach(split -> log.warn(
+ "split exceeds max configured split size len:{} max:{} extent:{} split:{}",
+ split.getLength(), maxSplitSize, extent, TextUtil.truncate(split)));
+
+ throw new ThriftTableOperationException(tableId.canonical(), null, tableOp,
+ TableOperationExceptionType.OTHER,
+ "Length of requested split exceeds tables configured max, see warning in logs for more information.");
+ }
+
+ goalMessage = "Splitting " + extent + " for user into " + (splits.size() + 1) + " tablets";
+ manager.fate(type).seedTransaction(op, fateId, new PreSplit(extent, splits), autoCleanup,
+ goalMessage);
+ break;
+ }
default:
throw new UnsupportedOperationException();
}
@@@ -888,7 -777,7 +887,7 @@@
// this is a bit redundant, the credentials of the caller (the first arg) will throw an
// exception if it fails to authenticate
// before the second arg is checked (which would return true or false)
-- if (!manager.security.authenticateUser(credentials, credentials)) {
++ if (!security.authenticateUser(credentials, credentials)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.BAD_CREDENTIALS);
}
@@@ -973,11 -861,10 +972,11 @@@
}
@Override
- public boolean cancelFateOperation(TInfo tinfo, TCredentials credentials, long opid)
+ public boolean cancelFateOperation(TInfo tinfo, TCredentials credentials, TFateId opid)
throws ThriftSecurityException, ThriftNotActiveServiceException {
+ FateId fateId = FateId.fromThrift(opid);
-- if (!manager.security.canPerformSystemActions(credentials)) {
++ if (!security.canPerformSystemActions(credentials)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
index e8fa02d89f,ee9314d958..fbf264350d
--- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
@@@ -145,7 -132,8 +145,6 @@@ import org.apache.accumulo.server.manag
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.security.SecurityOperation;
-import org.apache.accumulo.server.security.AuditedSecurityOperation;
import org.apache.accumulo.server.security.delegation.AuthenticationTokenKeyManager;
import org.apache.accumulo.server.security.delegation.ZooAuthenticationKeyDistributor;
import org.apache.accumulo.server.tables.TableManager;
@@@ -195,7 -182,7 +194,6 @@@ public class Manager extends AbstractSe
private final Object balancedNotifier = new Object();
final LiveTServerSet tserverSet;
private final List<TabletGroupWatcher> watchers = new ArrayList<>();
- final SecurityOperation security;
- final AuditedSecurityOperation security;
final Map<TServerInstance,AtomicInteger> badServers =
Collections.synchronizedMap(new HashMap<>());
final Set<TServerInstance> serversToShutdown = Collections.synchronizedSet(new HashSet<>());
@@@ -467,14 -427,10 +465,12 @@@
tserverSet = new LiveTServerSet(context, this);
initializeBalancer();
-- this.security = context.getSecurityOperation();
--
final long tokenLifetime = aconf.getTimeInMillis(Property.GENERAL_DELEGATION_TOKEN_LIFETIME);
+ this.rootTabletStore = TabletStateStore.getStoreForLevel(DataLevel.ROOT, context);
+ this.metadataTabletStore = TabletStateStore.getStoreForLevel(DataLevel.METADATA, context);
+ this.userTabletStore = TabletStateStore.getStoreForLevel(DataLevel.USER, context);
+
authenticationTokenKeyManager = null;
keyDistributor = null;
if (getConfiguration().getBoolean(Property.INSTANCE_RPC_SASL_ENABLED)) {
@@@ -1135,16 -1223,23 +1131,16 @@@
// Start the Manager's Fate Service
fateServiceHandler = new FateServiceHandler(this);
managerClientHandler = new ManagerClientServiceHandler(this);
- compactionCoordinator = new CompactionCoordinator(context, security, fateRefs, this);
- // Start the Manager's Client service
- // Ensure that calls before the manager gets the lock fail
- ManagerClientService.Iface haProxy =
- HighlyAvailableServiceWrapper.service(managerClientHandler, this);
++ compactionCoordinator = new CompactionCoordinator(this, fateRefs);
ServerAddress sa;
- var processor =
- ThriftProcessorTypes.getManagerTProcessor(this, fateServiceHandler, haProxy, getContext());
+ var processor = ThriftProcessorTypes.getManagerTProcessor(this, fateServiceHandler,
+ compactionCoordinator.getThriftService(), managerClientHandler, getContext());
try {
- @SuppressWarnings("deprecation")
- var maxMessageSizeProperty = getConfiguration().resolve(Property.RPC_MAX_MESSAGE_SIZE,
- Property.GENERAL_MAX_MESSAGE_SIZE);
- sa = TServerUtils.startServer(context, getHostname(), Property.MANAGER_CLIENTPORT, processor,
- "Manager", "Manager Client Service Handler", null, Property.MANAGER_MINTHREADS,
- Property.MANAGER_MINTHREADS_TIMEOUT, Property.MANAGER_THREADCHECK,
- maxMessageSizeProperty);
+ sa = TServerUtils.createThriftServer(context, getHostname(), Property.MANAGER_CLIENTPORT,
+ processor, "Manager", null, Property.MANAGER_MINTHREADS,
+ Property.MANAGER_MINTHREADS_TIMEOUT, Property.MANAGER_THREADCHECK);
} catch (UnknownHostException e) {
throw new IllegalStateException("Unable to start server on host " + getHostname(), e);
}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/ManagerClientServiceHandler.java
index de7eac7ce0,5a5b022e31..e5d7d5eee6
--- a/server/manager/src/main/java/org/apache/accumulo/manager/ManagerClientServiceHandler.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/ManagerClientServiceHandler.java
@@@ -81,10 -85,12 +81,12 @@@ import org.apache.accumulo.core.securit
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.manager.tableOps.TraceRepo;
import org.apache.accumulo.manager.tserverOps.ShutdownTServer;
++import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.client.ClientServiceHandler;
import org.apache.accumulo.server.conf.store.NamespacePropKey;
import org.apache.accumulo.server.conf.store.TablePropKey;
import org.apache.accumulo.server.manager.LiveTServerSet.TServerConnection;
-import org.apache.accumulo.server.replication.proto.Replication.Status;
++import org.apache.accumulo.server.security.AuditedSecurityOperation;
import org.apache.accumulo.server.security.delegation.AuthenticationTokenSecretManager;
import org.apache.accumulo.server.util.PropUtil;
import org.apache.accumulo.server.util.SystemPropUtil;
@@@ -99,10 -106,12 +101,14 @@@ import com.google.common.collect.Lists
public class ManagerClientServiceHandler implements ManagerClientService.Iface {
private static final Logger log = Manager.log;
- private static final Logger drainLog =
- LoggerFactory.getLogger("org.apache.accumulo.manager.ManagerDrainImpl");
private final Manager manager;
++ private final ServerContext context;
++ private final AuditedSecurityOperation security;
protected ManagerClientServiceHandler(Manager manager) {
this.manager = manager;
++ this.context = manager.getContext();
++ this.security = context.getSecurityOperation();
}
@Override
@@@ -110,13 -119,14 +116,13 @@@
throws ThriftSecurityException, ThriftTableOperationException {
TableId tableId = TableId.of(tableIdStr);
NamespaceId namespaceId = getNamespaceIdFromTableId(TableOperation.FLUSH, tableId);
-- if (!manager.security.canFlush(c, tableId, namespaceId)) {
++ if (!security.canFlush(c, tableId, namespaceId)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
- String zTablePath = Constants.ZROOT + "/" + manager.getInstanceID() + Constants.ZTABLES + "/"
- + tableId + Constants.ZTABLE_FLUSH_ID;
+ String zTablePath = Constants.ZTABLES + "/" + tableId + Constants.ZTABLE_FLUSH_ID;
- ZooReaderWriter zoo = manager.getContext().getZooSession().asReaderWriter();
- ZooReaderWriter zoo = manager.getContext().getZooReaderWriter();
++ ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
byte[] fid;
try {
fid = zoo.mutateExisting(zTablePath, currentValue -> {
@@@ -140,7 -150,7 +146,7 @@@
throws ThriftSecurityException, ThriftTableOperationException {
TableId tableId = TableId.of(tableIdStr);
NamespaceId namespaceId = getNamespaceIdFromTableId(TableOperation.FLUSH, tableId);
-- if (!manager.security.canFlush(c, tableId, namespaceId)) {
++ if (!security.canFlush(c, tableId, namespaceId)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -180,7 -190,7 +186,7 @@@
serversToFlush.clear();
-- try (TabletsMetadata tablets = TabletsMetadata.builder(manager.getContext()).forTable(tableId)
++ try (TabletsMetadata tablets = TabletsMetadata.builder(context).forTable(tableId)
.overlapping(startRow, endRow).fetch(FLUSH_ID, LOCATION, LOGS, PREV_ROW).build()) {
int tabletsToWaitFor = 0;
int tabletCount = 0;
@@@ -205,7 -215,7 +211,7 @@@
// TODO detect case of table offline AND tablets w/ logs? - ACCUMULO-1296
-- if (tabletCount == 0 && !manager.getContext().tableNodeExists(tableId)) {
++ if (tabletCount == 0 && !context.tableNodeExists(tableId)) {
throw new ThriftTableOperationException(tableId.canonical(), null, TableOperation.FLUSH,
TableOperationExceptionType.NOTFOUND, null);
}
@@@ -222,7 -232,7 +228,7 @@@
throws ThriftTableOperationException {
NamespaceId namespaceId;
try {
-- namespaceId = manager.getContext().getNamespaceId(tableId);
++ namespaceId = context.getNamespaceId(tableId);
} catch (TableNotFoundException e) {
throw new ThriftTableOperationException(tableId.canonical(), null, tableOp,
TableOperationExceptionType.NOTFOUND, e.getMessage());
@@@ -254,17 -264,18 +260,17 @@@
TVersionedProperties properties)
throws ThriftSecurityException, ThriftTableOperationException,
ThriftConcurrentModificationException, ThriftPropertyException {
-- final TableId tableId = ClientServiceHandler.checkTableId(manager.getContext(), tableName,
-- TableOperation.SET_PROPERTY);
++ final TableId tableId =
++ ClientServiceHandler.checkTableId(context, tableName, TableOperation.SET_PROPERTY);
NamespaceId namespaceId = getNamespaceIdFromTableId(TableOperation.SET_PROPERTY, tableId);
-- if (!manager.security.canAlterTable(credentials, tableId, namespaceId)) {
++ if (!security.canAlterTable(credentials, tableId, namespaceId)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
}
try {
- PropUtil.replaceProperties(manager.getContext(), TablePropKey.of(tableId),
- properties.getVersion(), properties.getProperties());
- PropUtil.replaceProperties(manager.getContext(),
- TablePropKey.of(manager.getContext(), tableId), properties.getVersion(),
++ PropUtil.replaceProperties(context, TablePropKey.of(tableId), properties.getVersion(),
+ properties.getProperties());
} catch (ConcurrentModificationException cme) {
log.warn("Error modifying table properties, properties have changed", cme);
throw new ThriftConcurrentModificationException(cme.getMessage());
@@@ -272,8 -283,8 +278,7 @@@
log.warn("Error modifying table properties: tableId: {}", tableId.canonical());
// race condition... table no longer exists? This call will throw an exception if the table
// was deleted:
-- ClientServiceHandler.checkTableId(manager.getContext(), tableName,
-- TableOperation.SET_PROPERTY);
++ ClientServiceHandler.checkTableId(context, tableName, TableOperation.SET_PROPERTY);
throw new ThriftTableOperationException(tableId.canonical(), tableName,
TableOperation.SET_PROPERTY, TableOperationExceptionType.OTHER,
"Error modifying table properties: tableId: " + tableId.canonical());
@@@ -286,7 -297,7 +291,7 @@@
@Override
public void shutdown(TInfo info, TCredentials c, boolean stopTabletServers)
throws ThriftSecurityException {
-- if (!manager.security.canPerformSystemActions(c)) {
++ if (!security.canPerformSystemActions(c)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
if (stopTabletServers) {
@@@ -302,7 -313,7 +307,7 @@@
@Override
public void shutdownTabletServer(TInfo info, TCredentials c, String tabletServer, boolean force)
throws ThriftSecurityException {
-- if (!manager.security.canPerformSystemActions(c)) {
++ if (!security.canPerformSystemActions(c)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -337,7 -346,7 +342,7 @@@
@Override
public void tabletServerStopping(TInfo tinfo, TCredentials credentials, String tabletServer)
throws ThriftSecurityException, ThriftNotActiveServiceException, TException {
-- if (!manager.security.canPerformSystemActions(credentials)) {
++ if (!security.canPerformSystemActions(credentials)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -348,7 -357,30 +353,7 @@@
@Override
public void reportTabletStatus(TInfo info, TCredentials credentials, String serverName,
TabletLoadState status, TKeyExtent ttablet) throws ThriftSecurityException {
-- if (!manager.security.canPerformSystemActions(credentials)) {
++ if (!security.canPerformSystemActions(credentials)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -380,7 -415,7 +385,7 @@@
@Override
public void setManagerGoalState(TInfo info, TCredentials c, ManagerGoalState state)
throws ThriftSecurityException {
-- if (!manager.security.canPerformSystemActions(c)) {
++ if (!security.canPerformSystemActions(c)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -390,12 -425,12 +395,12 @@@
@Override
public void removeSystemProperty(TInfo info, TCredentials c, String property)
throws ThriftSecurityException {
-- if (!manager.security.canPerformSystemActions(c)) {
++ if (!security.canPerformSystemActions(c)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
try {
-- SystemPropUtil.removeSystemProperty(manager.getContext(), property);
++ SystemPropUtil.removeSystemProperty(context, property);
updatePlugins(property);
} catch (Exception e) {
Manager.log.error("Problem removing config property in zookeeper", e);
@@@ -406,12 -441,12 +411,12 @@@
@Override
public void setSystemProperty(TInfo info, TCredentials c, String property, String value)
throws TException {
-- if (!manager.security.canPerformSystemActions(c)) {
++ if (!security.canPerformSystemActions(c)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
try {
-- SystemPropUtil.setSystemProperty(manager.getContext(), property, value);
++ SystemPropUtil.setSystemProperty(context, property, value);
updatePlugins(property);
} catch (IllegalArgumentException iae) {
Manager.log.error("Problem setting invalid property", iae);
@@@ -425,13 -460,13 +430,12 @@@
@Override
public void modifySystemProperties(TInfo info, TCredentials c, TVersionedProperties properties)
throws TException {
-- if (!manager.security.canPerformSystemActions(c)) {
++ if (!security.canPerformSystemActions(c)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
try {
-- SystemPropUtil.modifyProperties(manager.getContext(), properties.getVersion(),
-- properties.getProperties());
++ SystemPropUtil.modifyProperties(context, properties.getVersion(), properties.getProperties());
for (Map.Entry<String,String> entry : properties.getProperties().entrySet()) {
updatePlugins(entry.getKey());
}
@@@ -457,23 -492,24 +461,23 @@@
@Override
public void modifyNamespaceProperties(TInfo tinfo, TCredentials credentials, String ns,
TVersionedProperties properties) throws TException {
-- final NamespaceId namespaceId = ClientServiceHandler.checkNamespaceId(manager.getContext(), ns,
-- TableOperation.SET_PROPERTY);
-- if (!manager.security.canAlterNamespace(credentials, namespaceId)) {
++ final NamespaceId namespaceId =
++ ClientServiceHandler.checkNamespaceId(context, ns, TableOperation.SET_PROPERTY);
++ if (!security.canAlterNamespace(credentials, namespaceId)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
}
try {
- PropUtil.replaceProperties(manager.getContext(), NamespacePropKey.of(namespaceId),
- properties.getVersion(), properties.getProperties());
- PropUtil.replaceProperties(manager.getContext(),
- NamespacePropKey.of(manager.getContext(), namespaceId), properties.getVersion(),
++ PropUtil.replaceProperties(context, NamespacePropKey.of(namespaceId), properties.getVersion(),
+ properties.getProperties());
} catch (ConcurrentModificationException cme) {
log.warn("Error modifying namespace properties, properties have changed", cme);
throw new ThriftConcurrentModificationException(cme.getMessage());
} catch (IllegalStateException ex) {
// race condition on delete... namespace no longer exists? An undelying ZooKeeper.NoNode
// exception will be thrown an exception if the namespace was deleted:
-- ClientServiceHandler.checkNamespaceId(manager.getContext(), ns, TableOperation.SET_PROPERTY);
++ ClientServiceHandler.checkNamespaceId(context, ns, TableOperation.SET_PROPERTY);
log.warn("Error modifying namespace properties", ex);
throw new ThriftTableOperationException(namespaceId.canonical(), ns,
TableOperation.SET_PROPERTY, TableOperationExceptionType.OTHER,
@@@ -494,25 -530,25 +498,22 @@@
String value, TableOperation op)
throws ThriftSecurityException, ThriftTableOperationException, ThriftPropertyException {
-- NamespaceId namespaceId =
-- ClientServiceHandler.checkNamespaceId(manager.getContext(), namespace, op);
++ NamespaceId namespaceId = ClientServiceHandler.checkNamespaceId(context, namespace, op);
-- if (!manager.security.canAlterNamespace(c, namespaceId)) {
++ if (!security.canAlterNamespace(c, namespaceId)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
try {
if (value == null) {
- PropUtil.removeProperties(manager.getContext(), NamespacePropKey.of(namespaceId),
- List.of(property));
- PropUtil.removeProperties(manager.getContext(),
- NamespacePropKey.of(manager.getContext(), namespaceId), List.of(property));
++ PropUtil.removeProperties(context, NamespacePropKey.of(namespaceId), List.of(property));
} else {
- PropUtil.setProperties(manager.getContext(), NamespacePropKey.of(namespaceId),
- Map.of(property, value));
- PropUtil.setProperties(manager.getContext(),
- NamespacePropKey.of(manager.getContext(), namespaceId), Map.of(property, value));
++ PropUtil.setProperties(context, NamespacePropKey.of(namespaceId), Map.of(property, value));
}
} catch (IllegalStateException ex) {
// race condition on delete... namespace no longer exists? An undelying ZooKeeper.NoNode
// exception will be thrown an exception if the namespace was deleted:
-- ClientServiceHandler.checkNamespaceId(manager.getContext(), namespace, op);
++ ClientServiceHandler.checkNamespaceId(context, namespace, op);
log.info("Error altering namespace property", ex);
throw new ThriftTableOperationException(namespaceId.canonical(), namespace, op,
TableOperationExceptionType.OTHER, "Problem altering namespace property");
@@@ -524,29 -560,29 +525,27 @@@
private void alterTableProperty(TCredentials c, String tableName, String property, String value,
TableOperation op)
throws ThriftSecurityException, ThriftTableOperationException, ThriftPropertyException {
-- final TableId tableId = ClientServiceHandler.checkTableId(manager.getContext(), tableName, op);
++ final TableId tableId = ClientServiceHandler.checkTableId(context, tableName, op);
NamespaceId namespaceId = getNamespaceIdFromTableId(op, tableId);
-- if (!manager.security.canAlterTable(c, tableId, namespaceId)) {
++ if (!security.canAlterTable(c, tableId, namespaceId)) {
throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
}
try {
if (op == TableOperation.REMOVE_PROPERTY) {
- PropUtil.removeProperties(manager.getContext(), TablePropKey.of(tableId),
- List.of(property));
- PropUtil.removeProperties(manager.getContext(),
- TablePropKey.of(manager.getContext(), tableId), List.of(property));
++ PropUtil.removeProperties(context, TablePropKey.of(tableId), List.of(property));
} else if (op == TableOperation.SET_PROPERTY) {
if (value == null || value.isEmpty()) {
value = "";
}
- PropUtil.setProperties(manager.getContext(), TablePropKey.of(tableId),
- PropUtil.setProperties(manager.getContext(), TablePropKey.of(manager.getContext(), tableId),
-- Map.of(property, value));
++ PropUtil.setProperties(context, TablePropKey.of(tableId), Map.of(property, value));
}
} catch (IllegalStateException ex) {
log.warn("Invalid table property, tried to set: tableId: " + tableId.canonical() + " to: "
+ property + "=" + value);
// race condition... table no longer exists? This call will throw an exception if the table
// was deleted:
-- ClientServiceHandler.checkTableId(manager.getContext(), tableName, op);
++ ClientServiceHandler.checkTableId(context, tableName, op);
throw new ThriftTableOperationException(tableId.canonical(), tableName, op,
TableOperationExceptionType.OTHER, "Invalid table property, tried to set: tableId: "
+ tableId.canonical() + " to: " + property + "=" + value);
@@@ -572,7 -608,7 +571,7 @@@
@Override
public List<String> getActiveTservers(TInfo tinfo, TCredentials credentials)
throws ThriftSecurityException {
-- if (!manager.security.canPerformSystemActions(credentials)) {
++ if (!security.canPerformSystemActions(credentials)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -589,7 -625,7 +588,7 @@@
@Override
public TDelegationToken getDelegationToken(TInfo tinfo, TCredentials credentials,
TDelegationTokenConfig tConfig) throws ThriftSecurityException, TException {
-- if (!manager.security.canObtainDelegationToken(credentials)) {
++ if (!security.canObtainDelegationToken(credentials)) {
throw new ThriftSecurityException(credentials.getPrincipal(),
SecurityErrorCode.PERMISSION_DENIED);
}
@@@ -601,7 -637,7 +600,7 @@@
}
final DelegationTokenConfig config = DelegationTokenConfigSerializer.deserialize(tConfig);
-- final AuthenticationTokenSecretManager secretManager = manager.getContext().getSecretManager();
++ final AuthenticationTokenSecretManager secretManager = context.getSecretManager();
try {
Entry<Token<AuthenticationTokenIdentifier>,AuthenticationTokenIdentifier> pair =
secretManager.generateToken(credentials.principal, config);
@@@ -613,72 -649,107 +612,72 @@@
}
}
- @SuppressWarnings("deprecation")
@Override
- public boolean drainReplicationTable(TInfo tfino, TCredentials credentials, String tableName,
- Set<String> logsToWatch) throws TException {
- AccumuloClient client = manager.getContext();
+ public void requestTabletHosting(TInfo tinfo, TCredentials credentials, String tableIdStr,
+ List<TKeyExtent> extents) throws ThriftSecurityException, ThriftTableOperationException {
- final Text tableId = new Text(getTableId(manager.getContext(), tableName).canonical());
+ final TableId tableId = TableId.of(tableIdStr);
+ NamespaceId namespaceId = getNamespaceIdFromTableId(null, tableId);
- if (!manager.security.canScan(credentials, tableId, namespaceId)) {
++ if (!security.canScan(credentials, tableId, namespaceId)) {
+ throw new ThriftSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED);
+ }
- drainLog.trace("Waiting for {} to be replicated for {}", logsToWatch, tableId);
+ manager.mustBeOnline(tableId);
- drainLog.trace("Reading from metadata table");
- final Set<Range> range = Collections.singleton(new Range(ReplicationSection.getRange()));
- BatchScanner bs;
- try {
- bs = client.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 4);
- } catch (TableNotFoundException e) {
- throw new RuntimeException("Could not read metadata table", e);
- }
- bs.setRanges(range);
- bs.fetchColumnFamily(ReplicationSection.COLF);
- try {
- // Return immediately if there are records in metadata for these WALs
- if (!allReferencesReplicated(bs, tableId, logsToWatch)) {
- return false;
- }
- } finally {
- bs.close();
- }
+ manager.hostOndemand(Lists.transform(extents, KeyExtent::fromThrift));
+ }
- drainLog.trace("reading from replication table");
- try {
- bs = client.createBatchScanner(org.apache.accumulo.core.replication.ReplicationTable.NAME,
- Authorizations.EMPTY, 4);
- } catch (TableNotFoundException e) {
- throw new RuntimeException("Replication table was not found", e);
- }
- bs.setRanges(Collections.singleton(new Range()));
- try {
- // No records in metadata, check replication table
- return allReferencesReplicated(bs, tableId, logsToWatch);
- } finally {
- bs.close();
+ @Override
+ public List<TKeyExtent> updateTabletMergeability(TInfo tinfo, TCredentials credentials,
+ String tableName, Map<TKeyExtent,TTabletMergeability> splits)
+ throws ThriftTableOperationException, ThriftSecurityException {
+
- final TableId tableId = getTableId(manager.getContext(), tableName);
++ final TableId tableId = getTableId(context, tableName);
+ NamespaceId namespaceId = getNamespaceIdFromTableId(null, tableId);
+
- if (!manager.security.canSplitTablet(credentials, tableId, namespaceId)) {
++ if (!security.canSplitTablet(credentials, tableId, namespaceId)) {
+ throw new ThriftSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED);
}
- }
- try (var tabletsMutator = manager.getContext().getAmple().conditionallyMutateTablets()) {
- protected TableId getTableId(ClientContext context, String tableName)
- throws ThriftTableOperationException {
- return ClientServiceHandler.checkTableId(context, tableName, null);
- }
++ try (var tabletsMutator = context.getAmple().conditionallyMutateTablets()) {
+ for (Entry<TKeyExtent,TTabletMergeability> split : splits.entrySet()) {
+ var currentTime = manager.getSteadyTime();
+ var extent = KeyExtent.fromThrift(split.getKey());
+ var tabletMergeability = TabletMergeabilityUtil.fromThrift(split.getValue());
- /**
- * @return return true records are only in place which are fully replicated
- */
- @Deprecated
- protected boolean allReferencesReplicated(BatchScanner bs, Text tableId,
- Set<String> relevantLogs) {
- Text rowHolder = new Text(), colfHolder = new Text();
- for (Entry<Key,Value> entry : bs) {
- drainLog.trace("Got key {}", entry.getKey().toStringNoTruncate());
-
- entry.getKey().getColumnQualifier(rowHolder);
- if (tableId.equals(rowHolder)) {
- entry.getKey().getRow(rowHolder);
- entry.getKey().getColumnFamily(colfHolder);
-
- String file;
- if (colfHolder.equals(ReplicationSection.COLF)) {
- file = rowHolder.toString();
- file = file.substring(ReplicationSection.getRowPrefix().length());
- } else if (colfHolder
- .equals(org.apache.accumulo.core.replication.ReplicationSchema.OrderSection.NAME)) {
- file = org.apache.accumulo.core.replication.ReplicationSchema.OrderSection
- .getFile(entry.getKey(), rowHolder);
- long timeClosed = org.apache.accumulo.core.replication.ReplicationSchema.OrderSection
- .getTimeClosed(entry.getKey(), rowHolder);
- drainLog.trace("Order section: {} and {}", timeClosed, file);
- } else {
- file = rowHolder.toString();
- }
+ // Update the TabletMergeabilty metadata with the current manager time as
+ // long as there is no existing operation set
+ var updatedTm = TabletMergeabilityMetadata.toMetadata(tabletMergeability, currentTime);
+ tabletsMutator.mutateTablet(extent).requireAbsentOperation()
+ .putTabletMergeability(updatedTm)
+ .submit(tm -> updatedTm.equals(tm.getTabletMergeability()),
+ () -> "update TabletMergeability for " + extent + " to " + updatedTm);
+ }
- // Skip files that we didn't observe when we started (new files/data)
- if (relevantLogs.contains(file)) {
- drainLog.trace("Found file that we *do* care about {}", file);
+ var results = tabletsMutator.process();
+ List<TKeyExtent> updated = new ArrayList<>();
+ results.forEach((key, result) -> {
+ if (result.getStatus() == Status.ACCEPTED) {
+ updated.add(result.getExtent().toThrift());
} else {
- drainLog.trace("Found file that we didn't care about {}", file);
- continue;
+ log.debug("Unable to update TableMergeabilty: {}", result.getExtent());
}
-
- try {
- Status stat = Status.parseFrom(entry.getValue().get());
- if (!org.apache.accumulo.server.replication.StatusUtil.isFullyReplicated(stat)) {
- drainLog.trace("{} and {} is not replicated", file, ProtobufUtil.toString(stat));
- return false;
- }
- drainLog.trace("{} and {} is replicated", file, ProtobufUtil.toString(stat));
- } catch (InvalidProtocolBufferException e) {
- drainLog.trace("Could not parse protobuf for {}", entry.getKey(), e);
- }
- }
+ });
+ return updated;
}
+ }
- return true;
+ @Override
+ public long getManagerTimeNanos(TInfo tinfo, TCredentials credentials)
+ throws ThriftSecurityException {
- manager.security.authenticateUser(credentials, credentials);
++ security.authenticateUser(credentials, credentials);
+ return manager.getSteadyTime().getNanos();
+ }
+
+ protected TableId getTableId(ClientContext context, String tableName)
+ throws ThriftTableOperationException {
+ return ClientServiceHandler.checkTableId(context, tableName, null);
}
}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
index 6bc8edd89b,0000000000..6751b64d8a
mode 100644,000000..100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
@@@ -1,1320 -1,0 +1,1320 @@@
+/*
+ * 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
+ *
+ * https://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.accumulo.manager.compaction.coordinator;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.stream.Collectors.toList;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.COMPACTED;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.ECOMP;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.FILES;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.LOCATION;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.OPID;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.PREV_ROW;
+import static org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType.SELECTED;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.TableDeletedException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
+import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
+import org.apache.accumulo.core.client.admin.servers.ServerId;
+import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
+import org.apache.accumulo.core.clientImpl.thrift.TInfo;
+import org.apache.accumulo.core.clientImpl.thrift.TableOperation;
+import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
+import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException;
+import org.apache.accumulo.core.compaction.thrift.CompactionCoordinatorService;
+import org.apache.accumulo.core.compaction.thrift.TCompactionState;
+import org.apache.accumulo.core.compaction.thrift.TCompactionStatusUpdate;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompaction;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompactionList;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompactionMap;
+import org.apache.accumulo.core.compaction.thrift.TNextCompactionJob;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.NamespaceId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
+import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.FateId;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.FateKey;
+import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.iteratorsImpl.system.SystemIteratorUtil;
+import org.apache.accumulo.core.logging.TabletLogger;
+import org.apache.accumulo.core.manager.state.tables.TableState;
+import org.apache.accumulo.core.metadata.CompactableFileImpl;
+import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.schema.Ample;
+import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
+import org.apache.accumulo.core.metadata.schema.Ample.RejectionHandler;
+import org.apache.accumulo.core.metadata.schema.CompactionMetadata;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.metadata.schema.filters.HasExternalCompactionsFilter;
+import org.apache.accumulo.core.metrics.MetricsProducer;
+import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
+import org.apache.accumulo.core.spi.compaction.CompactionJob;
+import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.spi.compaction.CompactionPlanner;
+import org.apache.accumulo.core.spi.compaction.CompactionServiceId;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
+import org.apache.accumulo.core.tabletserver.thrift.InputFile;
+import org.apache.accumulo.core.tabletserver.thrift.IteratorConfig;
+import org.apache.accumulo.core.tabletserver.thrift.TCompactionKind;
+import org.apache.accumulo.core.tabletserver.thrift.TCompactionStats;
+import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.core.util.cache.Caches.CacheName;
+import org.apache.accumulo.core.util.compaction.CompactionPlannerInitParams;
+import org.apache.accumulo.core.util.compaction.CompactionServicesConfig;
+import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil;
+import org.apache.accumulo.core.util.compaction.RunningCompaction;
+import org.apache.accumulo.core.util.threads.ThreadPools;
+import org.apache.accumulo.core.util.threads.Threads;
+import org.apache.accumulo.core.volume.Volume;
+import org.apache.accumulo.manager.Manager;
+import org.apache.accumulo.manager.compaction.coordinator.commit.CommitCompaction;
+import org.apache.accumulo.manager.compaction.coordinator.commit.CompactionCommitData;
+import org.apache.accumulo.manager.compaction.coordinator.commit.RenameCompactionFile;
+import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue;
+import org.apache.accumulo.manager.compaction.queue.CompactionJobQueues;
+import org.apache.accumulo.manager.compaction.queue.ResolvedCompactionJob;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.ServiceEnvironmentImpl;
+import org.apache.accumulo.server.compaction.CompactionConfigStorage;
+import org.apache.accumulo.server.compaction.CompactionPluginUtils;
- import org.apache.accumulo.server.security.SecurityOperation;
++import org.apache.accumulo.server.security.AuditedSecurityOperation;
+import org.apache.accumulo.server.tablets.TabletNameGenerator;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.thrift.TException;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.github.benmanes.caffeine.cache.Weigher;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Suppliers;
+import com.google.common.collect.Sets;
+import com.google.common.net.HostAndPort;
+
+import io.micrometer.core.instrument.MeterRegistry;
+
+public class CompactionCoordinator
+ implements CompactionCoordinatorService.Iface, Runnable, MetricsProducer {
+
+ // Object that serves as a TopN view of the RunningCompactions, ordered by
+ // RunningCompaction start time. The first entry in this Set should be the
+ // oldest RunningCompaction.
+ public static class TimeOrderedRunningCompactionSet {
+
+ private static final int UPPER_LIMIT = 50;
+
+ Comparator<RunningCompaction> oldestFirstComparator =
+ Comparator.comparingLong(RunningCompaction::getStartTime)
+ .thenComparing(rc -> rc.getJob().getExternalCompactionId());
+ private final ConcurrentSkipListSet<RunningCompaction> compactions =
+ new ConcurrentSkipListSet<>(oldestFirstComparator);
+
+ // Tracking size here as ConcurrentSkipListSet.size() is not constant time
+ private final AtomicInteger size = new AtomicInteger(0);
+
+ public int size() {
+ return size.get();
+ }
+
+ public boolean add(RunningCompaction e) {
+ boolean added = compactions.add(e);
+ if (added) {
+ if (size.incrementAndGet() > UPPER_LIMIT) {
+ this.remove(compactions.last());
+ }
+ }
+ return added;
+ }
+
+ public boolean remove(Object o) {
+ boolean removed = compactions.remove(o);
+ if (removed) {
+ size.decrementAndGet();
+ }
+ return removed;
+ }
+
+ public Iterator<RunningCompaction> iterator() {
+ return compactions.iterator();
+ }
+
+ public Stream<RunningCompaction> stream() {
+ return compactions.stream();
+ }
+
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(CompactionCoordinator.class);
+
+ public static final String RESTART_UPDATE_MSG =
+ "Coordinator restarted, compaction found in progress";
+
+ /*
+ * Map of compactionId to RunningCompactions. This is an informational cache of what external
+ * compactions may be running. Its possible it may contain external compactions that are not
+ * actually running. It may not contain compactions that are actually running. The metadata table
+ * is the most authoritative source of what external compactions are currently running, but it
+ * does not have the stats that this map has.
+ */
+ protected final Map<ExternalCompactionId,RunningCompaction> RUNNING_CACHE =
+ new ConcurrentHashMap<>();
+
+ protected final Map<String,TimeOrderedRunningCompactionSet> LONG_RUNNING_COMPACTIONS_BY_RG =
+ new ConcurrentHashMap<>();
+
+ /* Map of group name to last time compactor called to get a compaction job */
+ private final Map<CompactorGroupId,Long> TIME_COMPACTOR_LAST_CHECKED = new ConcurrentHashMap<>();
+
+ private final ServerContext ctx;
- private final SecurityOperation security;
++ private final AuditedSecurityOperation security;
+ private final CompactionJobQueues jobQueues;
+ private final AtomicReference<Map<FateInstanceType,Fate<Manager>>> fateInstances;
+ // Exposed for tests
+ protected final CountDownLatch shutdown = new CountDownLatch(1);
+
+ private final ScheduledThreadPoolExecutor schedExecutor;
+
+ private final Cache<ExternalCompactionId,RunningCompaction> completed;
+ private final LoadingCache<FateId,CompactionConfig> compactionConfigCache;
+ private final Cache<Path,Integer> tabletDirCache;
+ private final DeadCompactionDetector deadCompactionDetector;
+
+ private final QueueMetrics queueMetrics;
+ private final Manager manager;
+
+ private final LoadingCache<String,Integer> compactorCounts;
+
+ private volatile long coordinatorStartTime;
+
+ private final Map<DataLevel,ThreadPoolExecutor> reservationPools;
+ private final Set<String> activeCompactorReservationRequest = ConcurrentHashMap.newKeySet();
+
- public CompactionCoordinator(ServerContext ctx, SecurityOperation security,
- AtomicReference<Map<FateInstanceType,Fate<Manager>>> fateInstances, Manager manager) {
- this.ctx = ctx;
++ public CompactionCoordinator(Manager manager,
++ AtomicReference<Map<FateInstanceType,Fate<Manager>>> fateInstances) {
++ this.ctx = manager.getContext();
+ this.schedExecutor = this.ctx.getScheduledExecutor();
- this.security = security;
++ this.security = ctx.getSecurityOperation();
+ this.manager = Objects.requireNonNull(manager);
+
+ long jobQueueMaxSize =
+ ctx.getConfiguration().getAsBytes(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE);
+
+ this.jobQueues = new CompactionJobQueues(jobQueueMaxSize);
+
+ this.queueMetrics = new QueueMetrics(jobQueues);
+
+ this.fateInstances = fateInstances;
+
+ completed = ctx.getCaches().createNewBuilder(CacheName.COMPACTIONS_COMPLETED, true)
+ .maximumSize(200).expireAfterWrite(10, TimeUnit.MINUTES).build();
+
+ CacheLoader<FateId,CompactionConfig> loader =
+ fateId -> CompactionConfigStorage.getConfig(ctx, fateId);
+
+ // Keep a small short lived cache of compaction config. Compaction config never changes, however
+ // when a compaction is canceled it is deleted which is why there is a time limit. It does not
+ // hurt to let a job that was canceled start, it will be canceled later. Caching this immutable
+ // config will help avoid reading the same data over and over.
+ compactionConfigCache = ctx.getCaches().createNewBuilder(CacheName.COMPACTION_CONFIGS, true)
+ .expireAfterWrite(30, SECONDS).maximumSize(100).build(loader);
+
+ Weigher<Path,Integer> weigher = (path, count) -> {
+ return path.toUri().toString().length();
+ };
+
+ tabletDirCache = ctx.getCaches().createNewBuilder(CacheName.COMPACTION_DIR_CACHE, true)
+ .maximumWeight(10485760L).weigher(weigher).build();
+
+ deadCompactionDetector =
+ new DeadCompactionDetector(this.ctx, this, schedExecutor, fateInstances);
+
+ var rootReservationPool = ThreadPools.getServerThreadPools().createExecutorService(
+ ctx.getConfiguration(), Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_ROOT, true);
+
+ var metaReservationPool = ThreadPools.getServerThreadPools().createExecutorService(
+ ctx.getConfiguration(), Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_META, true);
+
+ var userReservationPool = ThreadPools.getServerThreadPools().createExecutorService(
+ ctx.getConfiguration(), Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_USER, true);
+
+ reservationPools = Map.of(Ample.DataLevel.ROOT, rootReservationPool, Ample.DataLevel.METADATA,
+ metaReservationPool, Ample.DataLevel.USER, userReservationPool);
+
+ compactorCounts = ctx.getCaches().createNewBuilder(CacheName.COMPACTOR_COUNTS, false)
+ .expireAfterWrite(30, TimeUnit.SECONDS).build(this::countCompactors);
+ // At this point the manager does not have its lock so no actions should be taken yet
+ }
+
+ protected int countCompactors(String groupName) {
+ return ExternalCompactionUtil.countCompactors(groupName, ctx);
+ }
+
+ private volatile Thread serviceThread = null;
+
+ public void start() {
+ serviceThread = Threads.createThread("CompactionCoordinator Thread", this);
+ serviceThread.start();
+ }
+
+ public void shutdown() {
+ shutdown.countDown();
+
+ reservationPools.values().forEach(ExecutorService::shutdownNow);
+
+ var localThread = serviceThread;
+ if (localThread != null) {
+ try {
+ localThread.join();
+ } catch (InterruptedException e) {
+ LOG.error("Exception stopping compaction coordinator thread", e);
+ }
+ }
+ }
+
+ protected void startCompactorZKCleaner(ScheduledThreadPoolExecutor schedExecutor) {
+ ScheduledFuture<?> future = schedExecutor
+ .scheduleWithFixedDelay(this::cleanUpEmptyCompactorPathInZK, 0, 5, TimeUnit.MINUTES);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ }
+
+ protected void startInternalStateCleaner(ScheduledThreadPoolExecutor schedExecutor) {
+ ScheduledFuture<?> future =
+ schedExecutor.scheduleWithFixedDelay(this::cleanUpInternalState, 0, 5, TimeUnit.MINUTES);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ }
+
+ protected void startConfigMonitor(ScheduledThreadPoolExecutor schedExecutor) {
+ ScheduledFuture<?> future =
+ schedExecutor.scheduleWithFixedDelay(this::checkForConfigChanges, 0, 1, TimeUnit.MINUTES);
+ ThreadPools.watchNonCriticalScheduledTask(future);
+ }
+
+ private void checkForConfigChanges() {
+ long jobQueueMaxSize =
+ ctx.getConfiguration().getAsBytes(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE);
+ jobQueues.resetMaxSize(jobQueueMaxSize);
+ }
+
+ @Override
+ public void run() {
+
+ this.coordinatorStartTime = System.currentTimeMillis();
+ startConfigMonitor(schedExecutor);
+ startCompactorZKCleaner(schedExecutor);
+
+ // On a re-start of the coordinator it's possible that external compactions are in-progress.
+ // Attempt to get the running compactions on the compactors and then resolve which tserver
+ // the external compaction came from to re-populate the RUNNING collection.
+ LOG.info("Checking for running external compactions");
+ // On re-start contact the running Compactors to try and seed the list of running compactions
+ List<RunningCompaction> running = getCompactionsRunningOnCompactors();
+ if (running.isEmpty()) {
+ LOG.info("No running external compactions found");
+ } else {
+ LOG.info("Found {} running external compactions", running.size());
+ running.forEach(rc -> {
+ TCompactionStatusUpdate update = new TCompactionStatusUpdate();
+ update.setState(TCompactionState.IN_PROGRESS);
+ update.setMessage(RESTART_UPDATE_MSG);
+ rc.addUpdate(System.currentTimeMillis(), update);
+ rc.setStartTime(this.coordinatorStartTime);
+ RUNNING_CACHE.put(ExternalCompactionId.of(rc.getJob().getExternalCompactionId()), rc);
+ LONG_RUNNING_COMPACTIONS_BY_RG
+ .computeIfAbsent(rc.getGroupName(), k -> new TimeOrderedRunningCompactionSet()).add(rc);
+ });
+ }
+
+ startDeadCompactionDetector();
+ startInternalStateCleaner(schedExecutor);
+
+ try {
+ shutdown.await();
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted waiting for shutdown latch.", e);
+ }
+
+ LOG.info("Shutting down");
+ }
+
+ private Map<String,Set<HostAndPort>> getIdleCompactors(Set<ServerId> runningCompactors) {
+
+ final Map<String,Set<HostAndPort>> allCompactors = new HashMap<>();
+ runningCompactors.forEach(
+ (csi) -> allCompactors.computeIfAbsent(csi.getResourceGroup(), (k) -> new HashSet<>())
+ .add(HostAndPort.fromParts(csi.getHost(), csi.getPort())));
+
+ final Set<String> emptyQueues = new HashSet<>();
+
+ // Remove all of the compactors that are running a compaction
+ RUNNING_CACHE.values().forEach(rc -> {
+ Set<HostAndPort> busyCompactors = allCompactors.get(rc.getGroupName());
+ if (busyCompactors != null
+ && busyCompactors.remove(HostAndPort.fromString(rc.getCompactorAddress()))) {
+ if (busyCompactors.isEmpty()) {
+ emptyQueues.add(rc.getGroupName());
+ }
+ }
+ });
+ // Remove entries with empty queues
+ emptyQueues.forEach(e -> allCompactors.remove(e));
+ return allCompactors;
+ }
+
+ protected void startDeadCompactionDetector() {
+ deadCompactionDetector.start();
+ }
+
+ protected long getMissingCompactorWarningTime() {
+ return this.ctx.getConfiguration().getTimeInMillis(Property.COMPACTOR_MAX_JOB_WAIT_TIME) * 3;
+ }
+
+ public long getNumRunningCompactions() {
+ return RUNNING_CACHE.size();
+ }
+
+ /**
+ * Return the next compaction job from the queue to a Compactor
+ *
+ * @param groupName group
+ * @param compactorAddress compactor address
+ * @throws ThriftSecurityException when permission error
+ * @return compaction job
+ */
+ @Override
+ public TNextCompactionJob getCompactionJob(TInfo tinfo, TCredentials credentials,
+ String groupName, String compactorAddress, String externalCompactionId)
+ throws ThriftSecurityException {
+
+ // do not expect users to call this directly, expect compactors to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ CompactorGroupId groupId = CompactorGroupId.of(groupName);
+ LOG.trace("getCompactionJob called for group {} by compactor {}", groupId, compactorAddress);
+ TIME_COMPACTOR_LAST_CHECKED.put(groupId, System.currentTimeMillis());
+
+ TExternalCompactionJob result = null;
+
+ ResolvedCompactionJob rcJob = (ResolvedCompactionJob) jobQueues.poll(groupId);
+
+ while (rcJob != null) {
+
+ Optional<CompactionConfig> compactionConfig = getCompactionConfig(rcJob);
+
+ // this method may reread the metadata, do not use the metadata in rcJob for anything after
+ // this method
+ CompactionMetadata ecm = null;
+
+ var kind = rcJob.getKind();
+
+ // Only reserve user compactions when the config is present. When compactions are canceled the
+ // config is deleted.
+ var cid = ExternalCompactionId.from(externalCompactionId);
+ if (kind == CompactionKind.SYSTEM
+ || (kind == CompactionKind.USER && compactionConfig.isPresent())) {
+ ecm = reserveCompaction(rcJob, compactorAddress, cid);
+ }
+
+ if (ecm != null) {
+ result = createThriftJob(externalCompactionId, ecm, rcJob, compactionConfig);
+ // It is possible that by the time this added that the the compactor that made this request
+ // is dead. In this cases the compaction is not actually running.
+ RUNNING_CACHE.put(ExternalCompactionId.of(result.getExternalCompactionId()),
+ new RunningCompaction(result, compactorAddress, groupName));
+ TabletLogger.compacting(rcJob.getExtent(), rcJob.getSelectedFateId(), cid, compactorAddress,
+ rcJob);
+ break;
+ } else {
+ LOG.debug(
+ "Unable to reserve compaction job for {}, pulling another off the queue for group {}",
+ rcJob.getExtent(), groupName);
+ rcJob = (ResolvedCompactionJob) jobQueues.poll(CompactorGroupId.of(groupName));
+ }
+ }
+
+ if (rcJob == null) {
+ LOG.trace("No jobs found in group {} ", groupName);
+ }
+
+ if (result == null) {
+ LOG.trace("No jobs found for group {}, returning empty job to compactor {}", groupName,
+ compactorAddress);
+ result = new TExternalCompactionJob();
+ }
+
+ return new TNextCompactionJob(result, compactorCounts.get(groupName));
+ }
+
+ private void checkTabletDir(KeyExtent extent, Path path) {
+ try {
+ if (tabletDirCache.getIfPresent(path) == null) {
+ FileStatus[] files = null;
+ try {
+ files = ctx.getVolumeManager().listStatus(path);
+ } catch (FileNotFoundException ex) {
+ // ignored
+ }
+
+ if (files == null) {
+ LOG.debug("Tablet {} had no dir, creating {}", extent, path);
+
+ ctx.getVolumeManager().mkdirs(path);
+ }
+ tabletDirCache.put(path, 1);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ protected CompactionMetadata createExternalCompactionMetadata(ResolvedCompactionJob job,
+ String compactorAddress, ExternalCompactionId externalCompactionId) {
+ boolean propDels = !job.isCompactingAll();
+ FateId fateId = job.getSelectedFateId();
+
+ Consumer<String> directoryCreator = dir -> checkTabletDir(job.getExtent(), new Path(dir));
+ ReferencedTabletFile newFile = TabletNameGenerator.getNextDataFilenameForMajc(propDels, ctx,
+ job.getExtent(), job.getTabletDir(), directoryCreator, externalCompactionId);
+
+ return new CompactionMetadata(job.getJobFiles(), newFile, compactorAddress, job.getKind(),
+ job.getPriority(), job.getGroup(), propDels, fateId);
+
+ }
+
+ private class ReserveCompactionTask implements Supplier<CompactionMetadata> {
+ private final ResolvedCompactionJob rcJob;
+ private final String compactorAddress;
+ private final ExternalCompactionId externalCompactionId;
+
+ private ReserveCompactionTask(ResolvedCompactionJob rcJob, String compactorAddress,
+ ExternalCompactionId externalCompactionId) {
+ Preconditions.checkArgument(
+ rcJob.getKind() == CompactionKind.SYSTEM || rcJob.getKind() == CompactionKind.USER);
+ this.rcJob = Objects.requireNonNull(rcJob);
+ this.compactorAddress = Objects.requireNonNull(compactorAddress);
+ this.externalCompactionId = Objects.requireNonNull(externalCompactionId);
+ Preconditions.checkState(activeCompactorReservationRequest.add(compactorAddress),
+ "compactor %s already on has a reservation in flight, cannot process %s",
+ compactorAddress, externalCompactionId);
+ }
+
+ @Override
+ public CompactionMetadata get() {
+ if (ctx.getTableState(rcJob.getExtent().tableId()) != TableState.ONLINE) {
+ return null;
+ }
+
+ try {
+ try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) {
+ var extent = rcJob.getExtent();
+ var jobFiles = rcJob.getJobFiles();
+ long selectedExpirationDuration = ctx.getTableConfiguration(extent.tableId())
+ .getTimeInMillis(Property.TABLE_COMPACTION_SELECTION_EXPIRATION);
+ var reservationCheck = new CompactionReservationCheck(rcJob.getKind(), jobFiles,
+ rcJob.getSelectedFateId(), rcJob.isOverlapsSelectedFiles(), manager.getSteadyTime(),
+ selectedExpirationDuration);
+ var tabletMutator = tabletsMutator.mutateTablet(extent).requireAbsentOperation()
+ .requireCheckSuccess(reservationCheck);
+
+ var ecm = createExternalCompactionMetadata(rcJob, compactorAddress, externalCompactionId);
+
+ if (rcJob.isOverlapsSelectedFiles()) {
+ // There is corresponding code in CompactionReservationCheck that ensures this delete is
+ // safe to do.
+ tabletMutator.deleteSelectedFiles();
+ }
+ tabletMutator.putExternalCompaction(externalCompactionId, ecm);
+
+ tabletMutator.submit(tm -> tm.getExternalCompactions().containsKey(externalCompactionId),
+ () -> "compaction reservation");
+
+ var result = tabletsMutator.process().get(extent);
+
+ if (result.getStatus() == Ample.ConditionalResult.Status.ACCEPTED) {
+ return ecm;
+ } else {
+ return null;
+ }
+ }
+ } finally {
+ Preconditions.checkState(activeCompactorReservationRequest.remove(compactorAddress),
+ "compactorAddress:%s", compactorAddress);
+ }
+ }
+ }
+
+ protected CompactionMetadata reserveCompaction(ResolvedCompactionJob rcJob,
+ String compactorAddress, ExternalCompactionId externalCompactionId) {
+
+ if (activeCompactorReservationRequest.contains(compactorAddress)) {
+ // In this case the compactor has a previously submitted reservation request that is still
+ // processing. Do not want to let it queue up another reservation request. One possible cause
+ // of this is that compactor timed out waiting for its last request to process and is now
+ // making another request. The previously submitted request can not be used because the
+ // compactor generates a new uuid for each request it makes. So the best thing to do is to
+ // return null and wait for this situation to resolve. This will likely happen when some part
+ // of the distributed system is not working well, so at this point want to avoid making
+ // problems worse instead of trying to reserve a job.
+ LOG.warn(
+ "Ignoring request from {} to reserve compaction job because it has a reservation request in progress.",
+ compactorAddress);
+ return null;
+ }
+
+ var dataLevel = DataLevel.of(rcJob.getExtent().tableId());
+ var future = CompletableFuture.supplyAsync(
+ new ReserveCompactionTask(rcJob, compactorAddress, externalCompactionId),
+ reservationPools.get(dataLevel));
+ return future.join();
+ }
+
+ protected TExternalCompactionJob createThriftJob(String externalCompactionId,
+ CompactionMetadata ecm, ResolvedCompactionJob rcJob,
+ Optional<CompactionConfig> compactionConfig) {
+
+ // Only reach out to metadata table and get these if requested, usually not needed unless
+ // plugiun requests it.
+ Supplier<Set<CompactableFile>> selectedFiles = Suppliers.memoize(() -> {
+ if (rcJob.getKind() == CompactionKind.SYSTEM) {
+ return Set.of();
+ } else {
+ var tabletMetadata =
+ ctx.getAmple().readTablet(rcJob.getExtent(), SELECTED, FILES, PREV_ROW);
+ Preconditions.checkState(
+ tabletMetadata.getSelectedFiles().getFateId().equals(rcJob.getSelectedFateId()));
+ return tabletMetadata.getSelectedFiles().getFiles().stream()
+ .map(file -> new CompactableFileImpl(file, tabletMetadata.getFilesMap().get(file)))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+ });
+
+ Map<String,String> overrides = CompactionPluginUtils.computeOverrides(compactionConfig, ctx,
+ rcJob.getExtent(), rcJob.getFiles(), selectedFiles);
+
+ IteratorConfig iteratorSettings = SystemIteratorUtil
+ .toIteratorConfig(compactionConfig.map(CompactionConfig::getIterators).orElse(List.of()));
+
+ var files = rcJob.getJobFilesMap().entrySet().stream().map(e -> {
+ StoredTabletFile file = e.getKey();
+ DataFileValue dfv = e.getValue();
+ return new InputFile(file.getMetadata(), dfv.getSize(), dfv.getNumEntries(), dfv.getTime());
+ }).collect(toList());
+
+ // The fateId here corresponds to the Fate transaction that is driving a user initiated
+ // compaction. A system initiated compaction has no Fate transaction driving it so its ok to set
+ // it to null. If anything tries to use the id for a system compaction and triggers a NPE it's
+ // probably a bug that needs to be fixed.
+ FateId fateId = null;
+ if (rcJob.getKind() == CompactionKind.USER) {
+ fateId = rcJob.getSelectedFateId();
+ }
+
+ return new TExternalCompactionJob(externalCompactionId, rcJob.getExtent().toThrift(), files,
+ iteratorSettings, ecm.getCompactTmpName().getNormalizedPathStr(), ecm.getPropagateDeletes(),
+ TCompactionKind.valueOf(ecm.getKind().name()), fateId == null ? null : fateId.toThrift(),
+ overrides);
+ }
+
+ @Override
+ public void registerMetrics(MeterRegistry registry) {
+ queueMetrics.registerMetrics(registry);
+ }
+
+ public void addJobs(TabletMetadata tabletMetadata, Collection<CompactionJob> jobs) {
+ ArrayList<CompactionJob> resolvedJobs = new ArrayList<>(jobs.size());
+ for (var job : jobs) {
+ resolvedJobs.add(new ResolvedCompactionJob(job, tabletMetadata));
+ }
+
+ jobQueues.add(tabletMetadata.getExtent(), resolvedJobs);
+ }
+
+ public CompactionCoordinatorService.Iface getThriftService() {
+ return this;
+ }
+
+ private Optional<CompactionConfig> getCompactionConfig(ResolvedCompactionJob rcJob) {
+ if (rcJob.getKind() == CompactionKind.USER) {
+ var cconf = compactionConfigCache.get(rcJob.getSelectedFateId());
+ return Optional.ofNullable(cconf);
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Compactors calls this method when they have finished a compaction. This method does the
+ * following.
+ *
+ * <ol>
+ * <li>Reads the tablets metadata and determines if the compaction can commit. Its possible that
+ * things changed while the compaction was running and it can no longer commit.</li>
+ * <li>Commit the compaction using a conditional mutation. If the tablets files or location
+ * changed since reading the tablets metadata, then conditional mutation will fail. When this
+ * happens it will reread the metadata and go back to step 1 conceptually. When committing a
+ * compaction the compacted files are removed and scan entries are added to the tablet in case the
+ * files are in use, this prevents GC from deleting the files between updating tablet metadata and
+ * refreshing the tablet. The scan entries are only added when a tablet has a location.</li>
+ * <li>After successful commit a refresh request is sent to the tablet if it has a location. This
+ * will cause the tablet to start using the newly compacted files for future scans. Also the
+ * tablet can delete the scan entries if there are no active scans using them.</li>
+ * </ol>
+ *
+ * <p>
+ * User compactions will be refreshed as part of the fate operation. The user compaction fate
+ * operation will see the compaction was committed after this code updates the tablet metadata,
+ * however if it were to rely on this code to do the refresh it would not be able to know when the
+ * refresh was actually done. Therefore, user compactions will refresh as part of the fate
+ * operation so that it's known to be done before the fate operation returns. Since the fate
+ * operation will do it, there is no need to do it here for user compactions.
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @param externalCompactionId compaction id
+ * @param textent tablet extent
+ * @param stats compaction stats
+ * @throws ThriftSecurityException when permission error
+ */
+ @Override
+ public void compactionCompleted(TInfo tinfo, TCredentials credentials,
+ String externalCompactionId, TKeyExtent textent, TCompactionStats stats)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+
+ // maybe fate has not started yet
+ var localFates = fateInstances.get();
+ while (localFates == null) {
+ UtilWaitThread.sleep(100);
+ if (shutdown.getCount() == 0) {
+ return;
+ }
+ localFates = fateInstances.get();
+ }
+
+ var extent = KeyExtent.fromThrift(textent);
+ var localFate = localFates.get(FateInstanceType.fromTableId(extent.tableId()));
+
+ LOG.info("Compaction completed, id: {}, stats: {}, extent: {}", externalCompactionId, stats,
+ extent);
+ final var ecid = ExternalCompactionId.of(externalCompactionId);
+
+ var tabletMeta =
+ ctx.getAmple().readTablet(extent, ECOMP, SELECTED, LOCATION, FILES, COMPACTED, OPID);
+
+ var tableState = manager.getContext().getTableState(extent.tableId());
+ if (tableState != TableState.ONLINE) {
+ // Its important this check is done after the compaction id is set in the metadata table to
+ // avoid race conditions with the client code that waits for tables to go offline. That code
+ // looks for compaction ids in the metadata table after setting the table state. When that
+ // client code sees nothing for a tablet its important that nothing will changes the tablets
+ // files after that point in time which this check ensure.
+ LOG.debug("Not committing compaction {} for {} because of table state {}", ecid, extent,
+ tableState);
+ // cleanup metadata table and files related to the compaction
+ compactionsFailed(Map.of(ecid, extent));
+ return;
+ }
+
+ if (!CommitCompaction.canCommitCompaction(ecid, tabletMeta)) {
+ return;
+ }
+
+ // Start a fate transaction to commit the compaction.
+ CompactionMetadata ecm = tabletMeta.getExternalCompactions().get(ecid);
+ var renameOp = new RenameCompactionFile(new CompactionCommitData(ecid, extent, ecm, stats));
+ localFate.seedTransaction(Fate.FateOperation.COMMIT_COMPACTION,
+ FateKey.forCompactionCommit(ecid), renameOp, true);
+ }
+
+ @Override
+ public void compactionFailed(TInfo tinfo, TCredentials credentials, String externalCompactionId,
+ TKeyExtent extent) throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ KeyExtent fromThriftExtent = KeyExtent.fromThrift(extent);
+ LOG.info("Compaction failed, id: {}, extent: {}", externalCompactionId, fromThriftExtent);
+ final var ecid = ExternalCompactionId.of(externalCompactionId);
+ compactionsFailed(Map.of(ecid, KeyExtent.fromThrift(extent)));
+ }
+
+ void compactionsFailed(Map<ExternalCompactionId,KeyExtent> compactions) {
+ // Need to process each level by itself because the conditional tablet mutator does not support
+ // mutating multiple data levels at the same time. Also the conditional tablet mutator does not
+ // support submitting multiple mutations for a single tablet, so need to group by extent.
+
+ Map<DataLevel,Map<KeyExtent,Set<ExternalCompactionId>>> groupedCompactions =
+ new EnumMap<>(DataLevel.class);
+
+ compactions.forEach((ecid, extent) -> {
+ groupedCompactions.computeIfAbsent(DataLevel.of(extent.tableId()), dl -> new HashMap<>())
+ .computeIfAbsent(extent, e -> new HashSet<>()).add(ecid);
+ });
+
+ groupedCompactions
+ .forEach((dataLevel, levelCompactions) -> compactionFailedForLevel(levelCompactions));
+ }
+
+ void compactionFailedForLevel(Map<KeyExtent,Set<ExternalCompactionId>> compactions) {
+
+ try (var tabletsMutator = ctx.getAmple().conditionallyMutateTablets()) {
+ compactions.forEach((extent, ecids) -> {
+ try {
+ ctx.requireNotDeleted(extent.tableId());
+ var mutator = tabletsMutator.mutateTablet(extent).requireAbsentOperation();
+ ecids.forEach(mutator::requireCompaction);
+ ecids.forEach(mutator::deleteExternalCompaction);
+ mutator.submit(new RejectionHandler() {
+ @Override
+ public boolean callWhenTabletDoesNotExists() {
+ return true;
+ }
+
+ @Override
+ public boolean test(TabletMetadata tabletMetadata) {
+ return tabletMetadata == null
+ || Collections.disjoint(tabletMetadata.getExternalCompactions().keySet(), ecids);
+ }
+
+ });
+ } catch (TableDeletedException e) {
+ LOG.warn("Table {} was deleted, unable to update metadata for compaction failure.",
+ extent.tableId());
+ }
+ });
+
+ final List<ExternalCompactionId> ecidsForTablet = new ArrayList<>();
+ tabletsMutator.process().forEach((extent, result) -> {
+ if (result.getStatus() != Ample.ConditionalResult.Status.ACCEPTED) {
+
+ // this should try again later when the dead compaction detector runs, lets log it in case
+ // its a persistent problem
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Unable to remove failed compaction {} {}", extent, compactions.get(extent));
+ }
+ } else {
+ // compactionFailed is called from the Compactor when either a compaction fails or
+ // is cancelled and it's called from the DeadCompactionDetector. This block is
+ // entered when the conditional mutator above successfully deletes an ecid from
+ // the tablet metadata. Remove compaction tmp files from the tablet directory
+ // that have a corresponding ecid in the name.
+
+ ecidsForTablet.clear();
+ ecidsForTablet.addAll(compactions.get(extent));
+
+ if (!ecidsForTablet.isEmpty()) {
+ final TabletMetadata tm = ctx.getAmple().readTablet(extent, ColumnType.DIR);
+ if (tm != null) {
+ final Collection<Volume> vols = ctx.getVolumeManager().getVolumes();
+ for (Volume vol : vols) {
+ try {
+ final String volPath =
+ vol.getBasePath() + Constants.HDFS_TABLES_DIR + Path.SEPARATOR
+ + extent.tableId().canonical() + Path.SEPARATOR + tm.getDirName();
+ final FileSystem fs = vol.getFileSystem();
+ for (ExternalCompactionId ecid : ecidsForTablet) {
+ final String fileSuffix = "_tmp_" + ecid.canonical();
+ FileStatus[] files = null;
+ try {
+ files = fs.listStatus(new Path(volPath),
+ (path) -> path.getName().endsWith(fileSuffix));
+ } catch (FileNotFoundException e) {
+ LOG.trace("Failed to list tablet dir {}", volPath, e);
+ }
+ if (files != null) {
+ for (FileStatus file : files) {
+ if (!fs.delete(file.getPath(), false)) {
+ LOG.warn("Unable to delete ecid tmp file: {}: ", file.getPath());
+ } else {
+ LOG.debug("Deleted ecid tmp file: {}", file.getPath());
+ }
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOG.error("Exception deleting compaction tmp files for tablet: {}", extent, e);
+ }
+ }
+ } else {
+ // TabletMetadata does not exist for the extent. This could be due to a merge or
+ // split operation. Use the utility to find tmp files at the table level
+ deadCompactionDetector.addTableId(extent.tableId());
+ }
+ }
+ }
+ });
+ }
+
+ compactions.values().forEach(ecids -> ecids.forEach(this::recordCompletion));
+ }
+
+ /**
+ * Compactor calls to update the status of the assigned compaction
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @param externalCompactionId compaction id
+ * @param update compaction status update
+ * @param timestamp timestamp of the message
+ * @throws ThriftSecurityException when permission error
+ */
+ @Override
+ public void updateCompactionStatus(TInfo tinfo, TCredentials credentials,
+ String externalCompactionId, TCompactionStatusUpdate update, long timestamp)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ LOG.debug("Compaction status update, id: {}, timestamp: {}, update: {}", externalCompactionId,
+ timestamp, update);
+ final RunningCompaction rc = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId));
+ if (null != rc) {
+ rc.addUpdate(timestamp, update);
+ switch (update.state) {
+ case STARTED:
+ LONG_RUNNING_COMPACTIONS_BY_RG
+ .computeIfAbsent(rc.getGroupName(), k -> new TimeOrderedRunningCompactionSet())
+ .add(rc);
+ break;
+ case CANCELLED:
+ case FAILED:
+ case SUCCEEDED:
+ var compactionSet = LONG_RUNNING_COMPACTIONS_BY_RG.get(rc.getGroupName());
+ if (compactionSet != null) {
+ compactionSet.remove(rc);
+ }
+ break;
+ case ASSIGNED:
+ case IN_PROGRESS:
+ default:
+ // do nothing
+ break;
+
+ }
+ }
+ }
+
+ public void recordCompletion(ExternalCompactionId ecid) {
+ var rc = RUNNING_CACHE.remove(ecid);
+ if (rc != null) {
+ completed.put(ecid, rc);
+ var compactionSet = LONG_RUNNING_COMPACTIONS_BY_RG.get(rc.getGroupName());
+ if (compactionSet != null) {
+ compactionSet.remove(rc);
+ }
+ }
+ }
+
+ protected Set<ExternalCompactionId> readExternalCompactionIds() {
+ try (TabletsMetadata tabletsMetadata =
+ this.ctx.getAmple().readTablets().forLevel(Ample.DataLevel.USER)
+ .filter(new HasExternalCompactionsFilter()).fetch(ECOMP).build()) {
+ return tabletsMetadata.stream().flatMap(tm -> tm.getExternalCompactions().keySet().stream())
+ .collect(Collectors.toSet());
+ }
+ }
+
+ /**
+ * Return information about running compactions
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @return map of ECID to TExternalCompaction objects
+ * @throws ThriftSecurityException permission error
+ */
+ @Override
+ public TExternalCompactionMap getRunningCompactions(TInfo tinfo, TCredentials credentials)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+
+ final TExternalCompactionMap result = new TExternalCompactionMap();
+ RUNNING_CACHE.forEach((ecid, rc) -> {
+ TExternalCompaction trc = new TExternalCompaction();
+ trc.setGroupName(rc.getGroupName());
+ trc.setCompactor(rc.getCompactorAddress());
+ trc.setUpdates(rc.getUpdates());
+ trc.setJob(rc.getJob());
+ result.putToCompactions(ecid.canonical(), trc);
+ });
+ return result;
+ }
+
+ /**
+ * Return top 50 longest running compactions for each resource group
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @return map of group name to list of up to 50 compactions in sorted order, oldest compaction
+ * first.
+ * @throws ThriftSecurityException permission error
+ */
+ @Override
+ public Map<String,TExternalCompactionList> getLongRunningCompactions(TInfo tinfo,
+ TCredentials credentials) throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+
+ final Map<String,TExternalCompactionList> result = new HashMap<>();
+
+ for (Entry<String,TimeOrderedRunningCompactionSet> e : LONG_RUNNING_COMPACTIONS_BY_RG
+ .entrySet()) {
+ final TExternalCompactionList compactions = new TExternalCompactionList();
+ Iterator<RunningCompaction> iter = e.getValue().iterator();
+ while (iter.hasNext()) {
+ RunningCompaction rc = iter.next();
+ TExternalCompaction trc = new TExternalCompaction();
+ trc.setGroupName(rc.getGroupName());
+ trc.setCompactor(rc.getCompactorAddress());
+ trc.setUpdates(rc.getUpdates());
+ trc.setJob(rc.getJob());
+ compactions.addToCompactions(trc);
+ }
+ result.put(e.getKey(), compactions);
+ }
+ return result;
+ }
+
+ /**
+ * Return information about recently completed compactions
+ *
+ * @param tinfo trace info
+ * @param credentials tcredentials object
+ * @return map of ECID to TExternalCompaction objects
+ * @throws ThriftSecurityException permission error
+ */
+ @Override
+ public TExternalCompactionMap getCompletedCompactions(TInfo tinfo, TCredentials credentials)
+ throws ThriftSecurityException {
+ // do not expect users to call this directly, expect other tservers to call this method
+ if (!security.canPerformSystemActions(credentials)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ final TExternalCompactionMap result = new TExternalCompactionMap();
+ completed.asMap().forEach((ecid, rc) -> {
+ TExternalCompaction trc = new TExternalCompaction();
+ trc.setGroupName(rc.getGroupName());
+ trc.setCompactor(rc.getCompactorAddress());
+ trc.setJob(rc.getJob());
+ trc.setUpdates(rc.getUpdates());
+ result.putToCompactions(ecid.canonical(), trc);
+ });
+ return result;
+ }
+
+ @Override
+ public void cancel(TInfo tinfo, TCredentials credentials, String externalCompactionId)
+ throws TException {
+ var runningCompaction = RUNNING_CACHE.get(ExternalCompactionId.of(externalCompactionId));
+ var extent = KeyExtent.fromThrift(runningCompaction.getJob().getExtent());
+ try {
+ NamespaceId nsId = this.ctx.getNamespaceId(extent.tableId());
+ if (!security.canCompact(credentials, extent.tableId(), nsId)) {
+ throw new AccumuloSecurityException(credentials.getPrincipal(),
+ SecurityErrorCode.PERMISSION_DENIED).asThriftException();
+ }
+ } catch (TableNotFoundException e) {
+ throw new ThriftTableOperationException(extent.tableId().canonical(), null,
+ TableOperation.COMPACT_CANCEL, TableOperationExceptionType.NOTFOUND, e.getMessage());
+ }
+
+ cancelCompactionOnCompactor(runningCompaction.getCompactorAddress(), externalCompactionId);
+ }
+
+ /* Method exists to be called from test */
+ public CompactionJobQueues getJobQueues() {
+ return jobQueues;
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected List<RunningCompaction> getCompactionsRunningOnCompactors() {
+ return ExternalCompactionUtil.getCompactionsRunningOnCompactors(this.ctx);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected Set<ServerId> getRunningCompactors() {
+ return ctx.instanceOperations().getServers(ServerId.Type.COMPACTOR);
+ }
+
+ /* Method exists to be overridden in test to hide static method */
+ protected void cancelCompactionOnCompactor(String address, String externalCompactionId) {
+ HostAndPort hostPort = HostAndPort.fromString(address);
+ ExternalCompactionUtil.cancelCompaction(this.ctx, hostPort, externalCompactionId);
+ }
+
+ private void deleteEmpty(ZooReaderWriter zoorw, String path)
+ throws KeeperException, InterruptedException {
+ try {
+ LOG.debug("Deleting empty ZK node {}", path);
+ zoorw.delete(path);
+ } catch (KeeperException.NotEmptyException e) {
+ LOG.debug("Failed to delete {} its not empty, likely an expected race condition.", path);
+ }
+ }
+
+ private void cleanUpEmptyCompactorPathInZK() {
+
+ final var zoorw = this.ctx.getZooSession().asReaderWriter();
+
+ try {
+ var groups = zoorw.getChildren(Constants.ZCOMPACTORS);
+
+ for (String group : groups) {
+ final String qpath = Constants.ZCOMPACTORS + "/" + group;
+ final CompactorGroupId cgid = CompactorGroupId.of(group);
+ final var compactors = zoorw.getChildren(qpath);
+
+ if (compactors.isEmpty()) {
+ deleteEmpty(zoorw, qpath);
+ // Group has no compactors, we can clear its
+ // associated priority queue of jobs
+ CompactionJobPriorityQueue queue = getJobQueues().getQueue(cgid);
+ if (queue != null) {
+ queue.clearIfInactive(Duration.ofMinutes(10));
+ }
+ } else {
+ for (String compactor : compactors) {
+ String cpath = Constants.ZCOMPACTORS + "/" + group + "/" + compactor;
+ var lockNodes =
+ zoorw.getChildren(Constants.ZCOMPACTORS + "/" + group + "/" + compactor);
+ if (lockNodes.isEmpty()) {
+ deleteEmpty(zoorw, cpath);
+ }
+ }
+ }
+ }
+ } catch (KeeperException | RuntimeException e) {
+ LOG.warn("Failed to clean up compactors", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private Set<CompactorGroupId> getCompactionServicesConfigurationGroups()
+ throws ReflectiveOperationException, IllegalArgumentException, SecurityException {
+
+ Set<CompactorGroupId> groups = new HashSet<>();
+ AccumuloConfiguration config = ctx.getConfiguration();
+ CompactionServicesConfig servicesConfig = new CompactionServicesConfig(config);
+
+ for (var entry : servicesConfig.getPlanners().entrySet()) {
+ String serviceId = entry.getKey();
+ String plannerClassName = entry.getValue();
+
+ Class<? extends CompactionPlanner> plannerClass =
+ Class.forName(plannerClassName).asSubclass(CompactionPlanner.class);
+ CompactionPlanner planner = plannerClass.getDeclaredConstructor().newInstance();
+
+ var initParams = new CompactionPlannerInitParams(CompactionServiceId.of(serviceId),
+ servicesConfig.getPlannerPrefix(serviceId), servicesConfig.getOptions().get(serviceId),
+ new ServiceEnvironmentImpl(ctx));
+
+ planner.init(initParams);
+
+ groups.addAll(initParams.getRequestedGroups());
+ }
+ return groups;
+ }
+
+ public void cleanUpInternalState() {
+
+ // This method does the following:
+ //
+ // 1. Removes entries from RUNNING_CACHE and LONG_RUNNING_COMPACTIONS_BY_RG that are not really
+ // running
+ // 2. Cancels running compactions for groups that are not in the current configuration
+ // 3. Remove groups not in configuration from TIME_COMPACTOR_LAST_CHECKED
+ // 4. Log groups with no compactors
+ // 5. Log compactors with no groups
+ // 6. Log groups with compactors and queued jos that have not checked in
+
+ var config = ctx.getConfiguration();
+ ThreadPools.resizePool(reservationPools.get(DataLevel.ROOT), config,
+ Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_ROOT);
+ ThreadPools.resizePool(reservationPools.get(DataLevel.METADATA), config,
+ Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_META);
+ ThreadPools.resizePool(reservationPools.get(DataLevel.USER), config,
+ Property.COMPACTION_COORDINATOR_RESERVATION_THREADS_USER);
+
+ // grab a snapshot of the ids in the set before reading the metadata table. This is done to
+ // avoid removing things that are added while reading the metadata.
+ final Set<ExternalCompactionId> idsSnapshot = Set.copyOf(Sets.union(RUNNING_CACHE.keySet(),
+ LONG_RUNNING_COMPACTIONS_BY_RG.values().stream()
+ .flatMap(TimeOrderedRunningCompactionSet::stream)
+ .map(rc -> rc.getJob().getExternalCompactionId()).map(ExternalCompactionId::of)
+ .collect(Collectors.toSet())));
+
+ // grab the ids that are listed as running in the metadata table. It important that this is done
+ // after getting the snapshot.
+ final Set<ExternalCompactionId> idsInMetadata = readExternalCompactionIds();
+
+ final Set<ExternalCompactionId> idsToRemove = Sets.difference(idsSnapshot, idsInMetadata);
+
+ // remove ids that are in the running set but not in the metadata table
+ idsToRemove.forEach(this::recordCompletion);
+ if (idsToRemove.size() > 0) {
+ LOG.debug("Removed stale entries from RUNNING_CACHE : {}", idsToRemove);
+ }
+
+ // Get the set of groups being referenced in the current configuration
+ Set<CompactorGroupId> groupsInConfiguration = null;
+ try {
+ groupsInConfiguration = getCompactionServicesConfigurationGroups();
+ } catch (RuntimeException | ReflectiveOperationException e) {
+ LOG.error(
+ "Error getting groups from the compaction services configuration. Unable to clean up internal state.",
+ e);
+ return;
+ }
+
+ // Compaction jobs are created in the TabletGroupWatcher and added to the Coordinator
+ // via the addJobs method which adds the job to the CompactionJobQueues object.
+ final Set<CompactorGroupId> groupsWithJobs = jobQueues.getQueueIds();
+
+ final Set<CompactorGroupId> jobGroupsNotInConfiguration =
+ Sets.difference(groupsWithJobs, groupsInConfiguration);
+
+ if (jobGroupsNotInConfiguration != null && !jobGroupsNotInConfiguration.isEmpty()) {
+ RUNNING_CACHE.values().forEach(rc -> {
+ if (jobGroupsNotInConfiguration.contains(CompactorGroupId.of(rc.getGroupName()))) {
+ LOG.warn(
+ "External compaction {} running in group {} on compactor {},"
+ + " but group not found in current configuration. Failing compaction...",
+ rc.getJob().getExternalCompactionId(), rc.getGroupName(), rc.getCompactorAddress());
+ cancelCompactionOnCompactor(rc.getCompactorAddress(),
+ rc.getJob().getExternalCompactionId());
+ }
+ });
+
+ final Set<CompactorGroupId> trackedGroups = Set.copyOf(TIME_COMPACTOR_LAST_CHECKED.keySet());
+ TIME_COMPACTOR_LAST_CHECKED.keySet().retainAll(groupsInConfiguration);
+ LOG.debug("No longer tracking compactor check-in times for groups: {}",
+ Sets.difference(trackedGroups, TIME_COMPACTOR_LAST_CHECKED.keySet()));
+ }
+
+ final Set<ServerId> runningCompactors = getRunningCompactors();
+
+ final Set<CompactorGroupId> runningCompactorGroups = new HashSet<>();
+ runningCompactors
+ .forEach(c -> runningCompactorGroups.add(CompactorGroupId.of(c.getResourceGroup())));
+
+ final Set<CompactorGroupId> groupsWithNoCompactors =
+ Sets.difference(groupsInConfiguration, runningCompactorGroups);
+ if (groupsWithNoCompactors != null && !groupsWithNoCompactors.isEmpty()) {
+ for (CompactorGroupId group : groupsWithNoCompactors) {
+ long queuedJobCount = jobQueues.getQueuedJobs(group);
+ if (queuedJobCount > 0) {
+ LOG.warn("Compactor group {} has {} queued compactions but no running compactors", group,
+ queuedJobCount);
+ }
+ }
+ }
+
+ final Set<CompactorGroupId> compactorsWithNoGroups =
+ Sets.difference(runningCompactorGroups, groupsInConfiguration);
+ if (compactorsWithNoGroups != null && !compactorsWithNoGroups.isEmpty()) {
+ LOG.warn(
+ "The following groups have running compactors, but are not in the current configuration: {}",
+ compactorsWithNoGroups);
+ }
+
+ final long now = System.currentTimeMillis();
+ final long warningTime = getMissingCompactorWarningTime();
+ Map<String,Set<HostAndPort>> idleCompactors = getIdleCompactors(runningCompactors);
+ for (CompactorGroupId groupName : groupsInConfiguration) {
+ long lastCheckTime =
+ TIME_COMPACTOR_LAST_CHECKED.getOrDefault(groupName, coordinatorStartTime);
+ if ((now - lastCheckTime) > warningTime && jobQueues.getQueuedJobs(groupName) > 0
+ && idleCompactors.containsKey(groupName.canonical())) {
+ LOG.warn(
+ "The group {} has queued jobs and {} idle compactors, however none have checked in "
+ + "with coordinator for {}ms",
+ groupName, idleCompactors.get(groupName.canonical()).size(), warningTime);
+ }
+ }
+ }
+}
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/SetupPermissions.java
index 19787fb41d,49ff9f2914..ed8b30d3d2
--- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/SetupPermissions.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/SetupPermissions.java
@@@ -39,9 -37,9 +38,9 @@@ class SetupPermissions extends ManagerR
}
@Override
- public Repo<Manager> call(long tid, Manager env) throws Exception {
+ public Repo<Manager> call(FateId fateId, Manager env) throws Exception {
// give all table permissions to the creator
- SecurityOperation security = env.getContext().getSecurityOperation();
+ var security = env.getContext().getSecurityOperation();
if (!tableInfo.getUser().equals(env.getContext().getCredentials().getPrincipal())) {
for (TablePermission permission : TablePermission.values()) {
try {
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/tableOps/namespace/create/SetupNamespacePermissions.java
index e2aa1b7a49,9eac28dbfb..183438535e
--- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/namespace/create/SetupNamespacePermissions.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/namespace/create/SetupNamespacePermissions.java
@@@ -38,9 -36,9 +37,9 @@@ class SetupNamespacePermissions extend
}
@Override
- public Repo<Manager> call(long tid, Manager env) throws Exception {
+ public Repo<Manager> call(FateId fate, Manager env) throws Exception {
// give all namespace permissions to the creator
- SecurityOperation security = env.getContext().getSecurityOperation();
+ var security = env.getContext().getSecurityOperation();
for (var permission : NamespacePermission.values()) {
try {
security.grantNamespacePermission(env.getContext().rpcCreds(), namespaceInfo.user,
diff --cc server/manager/src/main/java/org/apache/accumulo/manager/tableOps/tableImport/ImportSetupPermissions.java
index f92be7c1fa,4f622a78e9..402cbcb046
--- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/tableImport/ImportSetupPermissions.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/tableImport/ImportSetupPermissions.java
@@@ -43,9 -41,9 +42,9 @@@ class ImportSetupPermissions extends Ma
}
@Override
- public Repo<Manager> call(long tid, Manager env) throws Exception {
+ public Repo<Manager> call(FateId fateId, Manager env) throws Exception {
// give all table permissions to the creator
- SecurityOperation security = env.getContext().getSecurityOperation();
+ var security = env.getContext().getSecurityOperation();
for (TablePermission permission : TablePermission.values()) {
try {
security.grantTablePermission(env.getContext().rpcCreds(), tableInfo.user,
diff --cc server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java
index 4035015213,0000000000..e254634167
mode 100644,000000..100644
--- a/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java
+++ b/server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java
@@@ -1,458 -1,0 +1,413 @@@
+/*
+ * 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
+ *
+ * https://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.accumulo.manager.compaction;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
++import static org.easymock.EasyMock.verify;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.admin.CompactionConfig;
+import org.apache.accumulo.core.client.admin.servers.ServerId;
+import org.apache.accumulo.core.clientImpl.thrift.TInfo;
+import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.compaction.thrift.TExternalCompaction;
+import org.apache.accumulo.core.compaction.thrift.TNextCompactionJob;
+import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.dataImpl.thrift.TKeyExtent;
+import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.FateId;
+import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.iteratorsImpl.system.SystemIteratorUtil;
+import org.apache.accumulo.core.metadata.ReferencedTabletFile;
+import org.apache.accumulo.core.metadata.schema.CompactionMetadata;
+import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metrics.MetricsInfo;
+import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
+import org.apache.accumulo.core.spi.compaction.CompactionJob;
+import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
+import org.apache.accumulo.core.tabletserver.thrift.TCompactionKind;
+import org.apache.accumulo.core.tabletserver.thrift.TCompactionStats;
+import org.apache.accumulo.core.tabletserver.thrift.TExternalCompactionJob;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.core.util.cache.Caches;
+import org.apache.accumulo.core.util.compaction.CompactionJobImpl;
+import org.apache.accumulo.core.util.compaction.RunningCompaction;
+import org.apache.accumulo.core.util.time.SteadyTime;
+import org.apache.accumulo.manager.Manager;
+import org.apache.accumulo.manager.compaction.coordinator.CompactionCoordinator;
+import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue;
+import org.apache.accumulo.manager.compaction.queue.ResolvedCompactionJob;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.conf.TableConfiguration;
+import org.apache.accumulo.server.security.AuditedSecurityOperation;
- import org.apache.accumulo.server.security.SecurityOperation;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
- import org.easymock.EasyMock;
++import org.junit.jupiter.api.AfterEach;
++import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
++import org.junit.jupiter.api.TestInfo;
+
+import com.google.common.net.HostAndPort;
+
+public class CompactionCoordinatorTest {
+
+ // Need a non-null fateInstances reference for CompactionCoordinator.compactionCompleted
+ private static final AtomicReference<Map<FateInstanceType,Fate<Manager>>> fateInstances =
+ new AtomicReference<>(Map.of());
+
+ private static final CompactorGroupId GROUP_ID = CompactorGroupId.of("R2DQ");
+
+ private final HostAndPort tserverAddr = HostAndPort.fromParts("192.168.1.1", 9090);
+
+ public MetricsInfo getMockMetrics() {
+ MetricsInfo metricsInfo = createMock(MetricsInfo.class);
+ metricsInfo.addMetricsProducers(anyObject());
+ expectLastCall().anyTimes();
+ metricsInfo.init(List.of());
+ expectLastCall().anyTimes();
- replay(metricsInfo);
+ return metricsInfo;
+ }
+
+ public class TestCoordinator extends CompactionCoordinator {
+
+ private final List<RunningCompaction> runningCompactions;
+
+ private Set<ExternalCompactionId> metadataCompactionIds = null;
+
- public TestCoordinator(ServerContext ctx, SecurityOperation security,
- List<RunningCompaction> runningCompactions, Manager manager) {
- super(ctx, security, fateInstances, manager);
++ public TestCoordinator(Manager manager, List<RunningCompaction> runningCompactions) {
++ super(manager, fateInstances);
+ this.runningCompactions = runningCompactions;
+ }
+
+ @Override
+ protected int countCompactors(String groupName) {
+ return 3;
+ }
+
+ @Override
+ protected void startDeadCompactionDetector() {}
+
+ @Override
+ protected void startCompactorZKCleaner(ScheduledThreadPoolExecutor schedExecutor) {}
+
+ @Override
+ protected void startInternalStateCleaner(ScheduledThreadPoolExecutor schedExecutor) {
+ // This is called from CompactionCoordinator.run(). Counting down
+ // the latch will exit the run method
+ this.shutdown.countDown();
+ }
+
+ @Override
+ protected void startConfigMonitor(ScheduledThreadPoolExecutor schedExecutor) {}
+
+ @Override
+ public void compactionCompleted(TInfo tinfo, TCredentials credentials,
+ String externalCompactionId, TKeyExtent textent, TCompactionStats stats)
+ throws ThriftSecurityException {}
+
+ @Override
+ public void compactionFailed(TInfo tinfo, TCredentials credentials, String externalCompactionId,
+ TKeyExtent extent) throws ThriftSecurityException {}
+
+ void setMetadataCompactionIds(Set<ExternalCompactionId> mci) {
+ metadataCompactionIds = mci;
+ }
+
+ @Override
+ protected Set<ExternalCompactionId> readExternalCompactionIds() {
+ if (metadataCompactionIds == null) {
+ return RUNNING_CACHE.keySet();
+ } else {
+ return metadataCompactionIds;
+ }
+ }
+
+ public Map<ExternalCompactionId,RunningCompaction> getRunning() {
+ return RUNNING_CACHE;
+ }
+
+ public Map<String,TimeOrderedRunningCompactionSet> getLongRunningByGroup() {
+ return LONG_RUNNING_COMPACTIONS_BY_RG;
+ }
+
+ public void resetInternals() {
+ getRunning().clear();
+ metadataCompactionIds = null;
+ }
+
+ @Override
+ protected List<RunningCompaction> getCompactionsRunningOnCompactors() {
+ return runningCompactions;
+ }
+
+ @Override
+ protected Set<ServerId> getRunningCompactors() {
+ return Set.of();
+ }
+
+ @Override
+ protected CompactionMetadata reserveCompaction(ResolvedCompactionJob rcJob,
+ String compactorAddress, ExternalCompactionId externalCompactionId) {
+ return createExternalCompactionMetadata(rcJob, compactorAddress, externalCompactionId);
+ }
+
+ @Override
+ protected CompactionMetadata createExternalCompactionMetadata(ResolvedCompactionJob job,
+ String compactorAddress, ExternalCompactionId externalCompactionId) {
+ FateInstanceType type = FateInstanceType.fromTableId(job.getExtent().tableId());
+ FateId fateId = FateId.from(type, UUID.randomUUID());
+ return new CompactionMetadata(job.getJobFiles(),
+ new ReferencedTabletFile(new Path("file:///accumulo/tables/1/default_tablet/F00001.rf")),
+ compactorAddress, job.getKind(), job.getPriority(), job.getGroup(), true, fateId);
+ }
+
+ @Override
+ protected TExternalCompactionJob createThriftJob(String externalCompactionId,
+ CompactionMetadata ecm, ResolvedCompactionJob rcJob,
+ Optional<CompactionConfig> compactionConfig) {
+ return new TExternalCompactionJob(externalCompactionId, rcJob.getExtent().toThrift(),
+ List.of(), SystemIteratorUtil.toIteratorConfig(List.of()),
+ ecm.getCompactTmpName().getNormalizedPathStr(), ecm.getPropagateDeletes(),
+ TCompactionKind.valueOf(ecm.getKind().name()),
+ FateId.from(FateInstanceType.fromTableId(rcJob.getExtent().tableId()), UUID.randomUUID())
+ .toThrift(),
+ Map.of());
+ }
+
+ @Override
+ protected void cancelCompactionOnCompactor(String address, String externalCompactionId) {}
+
+ }
+
- @Test
- public void testCoordinatorColdStart() throws Exception {
++ private TableId tableId;
++ private Manager manager;
++ private ServerContext context;
++ private AuditedSecurityOperation security;
++ private MetricsInfo metricsInfo;
++ private TableConfiguration tconf;
++ private TCredentials rpcCreds;
+
- ServerContext context = EasyMock.createNiceMock(ServerContext.class);
- expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes();
- expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
++ @BeforeEach
++ public void setupMocks(TestInfo testInfo) throws Exception {
++ rpcCreds = new TCredentials(null, null, null, null);
+
- MetricsInfo metricsInfo = getMockMetrics();
- expect(context.getMetricsInfo()).andReturn(metricsInfo).anyTimes();
++ tableId = TableId.of(testInfo.getDisplayName());
+
- AuditedSecurityOperation security = EasyMock.createNiceMock(AuditedSecurityOperation.class);
++ metricsInfo = getMockMetrics();
++
++ security = createMock(AuditedSecurityOperation.class);
++ expect(security.canPerformSystemActions(rpcCreds)).andReturn(true).anyTimes();
++
++ context = createMock(ServerContext.class);
++ expect(context.getMetricsInfo()).andReturn(getMockMetrics()).anyTimes();
++ expect(context.getSecurityOperation()).andReturn(security).anyTimes();
++ expect(context.getScheduledExecutor()).andReturn(null).anyTimes();
++ expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes();
++ expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
++ tconf = createMock(TableConfiguration.class);
++ expect(tconf.get(Property.TABLE_COMPACTION_CONFIGURER))
++ .andReturn(Property.TABLE_COMPACTION_CONFIGURER.getDefaultValue()).anyTimes();
++ expect(context.getTableConfiguration(tableId)).andReturn(tconf).anyTimes();
+
- Manager manager = EasyMock.createNiceMock(Manager.class);
++ manager = createMock(Manager.class);
++ expect(manager.getContext()).andReturn(context).anyTimes();
+ expect(manager.getSteadyTime()).andReturn(SteadyTime.from(100000, TimeUnit.NANOSECONDS))
+ .anyTimes();
+
- EasyMock.replay(context, security, manager);
++ replay(manager, context, security, metricsInfo, tconf);
++ }
++
++ @AfterEach
++ public void verifyMocks() {
++ verify(manager, context, security, metricsInfo, tconf);
++ }
+
- var coordinator = new TestCoordinator(context, security, new ArrayList<>(), manager);
++ @Test
++ public void testCoordinatorColdStart() throws Exception {
++ var coordinator = new TestCoordinator(manager, new ArrayList<>());
+ assertEquals(0, coordinator.getJobQueues().getQueuedJobCount());
+ assertEquals(0, coordinator.getRunning().size());
+ assertEquals(0, coordinator.getLongRunningByGroup().size());
+ coordinator.run();
+ coordinator.shutdown();
+
+ assertEquals(0, coordinator.getJobQueues().getQueuedJobCount());
+ assertEquals(0, coordinator.getRunning().size());
+ assertEquals(0, coordinator.getLongRunningByGroup().size());
- EasyMock.verify(context, security, metricsInfo);
+ }
+
+ @Test
+ public void testCoordinatorRestartOneRunningCompaction() throws Exception {
-
- ServerContext context = EasyMock.createNiceMock(ServerContext.class);
- expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes();
- expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
-
- MetricsInfo metricsInfo = getMockMetrics();
- expect(context.getMetricsInfo()).andReturn(metricsInfo).anyTimes();
-
+ List<RunningCompaction> runningCompactions = new ArrayList<>();
+ ExternalCompactionId eci = ExternalCompactionId.generate(UUID.randomUUID());
- TExternalCompactionJob job = EasyMock.createNiceMock(TExternalCompactionJob.class);
- expect(job.getExternalCompactionId()).andReturn(eci.toString()).anyTimes();
++
++ TExternalCompactionJob job = createMock(TExternalCompactionJob.class);
++ expect(job.getExternalCompactionId()).andReturn(eci.toString()).atLeastOnce();
+ TKeyExtent extent = new TKeyExtent();
+ extent.setTable("1".getBytes(UTF_8));
+ runningCompactions.add(new RunningCompaction(job, tserverAddr.toString(), GROUP_ID.toString()));
++ replay(job);
+
- AuditedSecurityOperation security = EasyMock.createNiceMock(AuditedSecurityOperation.class);
-
- Manager manager = EasyMock.createNiceMock(Manager.class);
- expect(manager.getSteadyTime()).andReturn(SteadyTime.from(100000, TimeUnit.NANOSECONDS))
- .anyTimes();
-
- EasyMock.replay(context, job, security, manager);
-
- var coordinator = new TestCoordinator(context, security, runningCompactions, manager);
++ var coordinator = new TestCoordinator(manager, runningCompactions);
+ coordinator.resetInternals();
+ assertEquals(0, coordinator.getJobQueues().getQueuedJobCount());
+ assertEquals(0, coordinator.getRunning().size());
+ assertEquals(0, coordinator.getLongRunningByGroup().size());
+ coordinator.run();
+ coordinator.shutdown();
+ assertEquals(0, coordinator.getJobQueues().getQueuedJobCount());
+ assertEquals(1, coordinator.getRunning().size());
+ assertEquals(1, coordinator.getLongRunningByGroup().size());
+
+ Map<ExternalCompactionId,RunningCompaction> running = coordinator.getRunning();
+ Entry<ExternalCompactionId,RunningCompaction> ecomp = running.entrySet().iterator().next();
+ assertEquals(eci, ecomp.getKey());
+ RunningCompaction rc = ecomp.getValue();
+ assertEquals(GROUP_ID.toString(), rc.getGroupName());
+ assertEquals(tserverAddr.toString(), rc.getCompactorAddress());
+
+ assertTrue(coordinator.getLongRunningByGroup().containsKey(GROUP_ID.toString()));
+ assertTrue(coordinator.getLongRunningByGroup().get(GROUP_ID.toString()).size() == 1);
+ rc = coordinator.getLongRunningByGroup().get(GROUP_ID.toString()).iterator().next();
+ assertEquals(GROUP_ID.toString(), rc.getGroupName());
+ assertEquals(tserverAddr.toString(), rc.getCompactorAddress());
+
- EasyMock.verify(context, job, security);
++ verify(job);
+ }
+
+ @Test
+ public void testGetCompactionJob() throws Exception {
++ KeyExtent ke = new KeyExtent(tableId, new Text("z"), new Text("b"));
+
- TableConfiguration tconf = EasyMock.createNiceMock(TableConfiguration.class);
- expect(tconf.get(Property.TABLE_COMPACTION_CONFIGURER))
- .andReturn(Property.TABLE_COMPACTION_CONFIGURER.getDefaultValue()).anyTimes();
-
- ServerContext context = EasyMock.createNiceMock(ServerContext.class);
- expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes();
- expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
- expect(context.getTableConfiguration(TableId.of("2a"))).andReturn(tconf).anyTimes();
-
- MetricsInfo metricsInfo = getMockMetrics();
- expect(context.getMetricsInfo()).andReturn(metricsInfo).anyTimes();
-
- TCredentials creds = EasyMock.createNiceMock(TCredentials.class);
- expect(context.rpcCreds()).andReturn(creds).anyTimes();
-
- AuditedSecurityOperation security = EasyMock.createNiceMock(AuditedSecurityOperation.class);
- expect(security.canPerformSystemActions(creds)).andReturn(true).anyTimes();
-
- KeyExtent ke = new KeyExtent(TableId.of("2a"), new Text("z"), new Text("b"));
- TabletMetadata tm = EasyMock.createNiceMock(TabletMetadata.class);
- expect(tm.getExtent()).andReturn(ke).anyTimes();
- expect(tm.getFiles()).andReturn(Collections.emptySet()).anyTimes();
- expect(tm.getTableId()).andReturn(ke.tableId()).anyTimes();
- expect(tm.getDirName()).andReturn("t-00001").anyTimes();
- Manager manager = EasyMock.createNiceMock(Manager.class);
- expect(manager.getSteadyTime()).andReturn(SteadyTime.from(100000, TimeUnit.NANOSECONDS))
- .anyTimes();
++ TabletMetadata tm = createMock(TabletMetadata.class);
++ expect(tm.getSelectedFiles()).andReturn(null).atLeastOnce();
++ expect(tm.getExtent()).andReturn(ke).atLeastOnce();
++ expect(tm.getFiles()).andReturn(Collections.emptySet()).atLeastOnce();
++ expect(tm.getDirName()).andReturn("t-00001").atLeastOnce();
++ replay(tm);
+
- EasyMock.replay(tconf, context, creds, tm, security, manager);
-
- var coordinator = new TestCoordinator(context, security, new ArrayList<>(), manager);
++ var coordinator = new TestCoordinator(manager, new ArrayList<>());
+ assertEquals(0, coordinator.getJobQueues().getQueuedJobCount());
+ assertEquals(0, coordinator.getRunning().size());
+ // Use coordinator.run() to populate the internal data structures. This is tested in a different
+ // test.
+ coordinator.run();
+ coordinator.shutdown();
+
+ assertEquals(0, coordinator.getJobQueues().getQueuedJobCount());
+ assertEquals(0, coordinator.getRunning().size());
+
+ // Add a job to the job queue
+ CompactionJob job =
+ new CompactionJobImpl((short) 1, GROUP_ID, Collections.emptyList(), CompactionKind.SYSTEM);
+ coordinator.addJobs(tm, Collections.singleton(job));
+ CompactionJobPriorityQueue queue = coordinator.getJobQueues().getQueue(GROUP_ID);
+ assertEquals(1, queue.getQueuedJobs());
+
+ // Get the next job
+ ExternalCompactionId eci = ExternalCompactionId.generate(UUID.randomUUID());
- TNextCompactionJob nextJob = coordinator.getCompactionJob(new TInfo(), creds,
++ TNextCompactionJob nextJob = coordinator.getCompactionJob(new TInfo(), rpcCreds,
+ GROUP_ID.toString(), "localhost:10241", eci.toString());
+ assertEquals(3, nextJob.getCompactorCount());
+ TExternalCompactionJob createdJob = nextJob.getJob();
+ assertEquals(eci.toString(), createdJob.getExternalCompactionId());
+ assertEquals(ke, KeyExtent.fromThrift(createdJob.getExtent()));
+
+ assertEquals(0, coordinator.getJobQueues().getQueuedJobCount());
+ assertEquals(1, coordinator.getRunning().size());
+ Entry<ExternalCompactionId,RunningCompaction> entry =
+ coordinator.getRunning().entrySet().iterator().next();
+ assertEquals(eci.toString(), entry.getKey().toString());
+ assertEquals("localhost:10241", entry.getValue().getCompactorAddress());
+ assertEquals(eci.toString(), entry.getValue().getJob().getExternalCompactionId());
+
- EasyMock.verify(tconf, context, creds, tm, security);
++ verify(tm);
+ }
+
+ @Test
+ public void testGetCompactionJobNoJobs() throws Exception {
-
- ServerContext context = EasyMock.createNiceMock(ServerContext.class);
- expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes();
- expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
-
- TCredentials creds = EasyMock.createNiceMock(TCredentials.class);
-
- AuditedSecurityOperation security = EasyMock.createNiceMock(AuditedSecurityOperation.class);
- expect(security.canPerformSystemActions(creds)).andReturn(true);
-
- Manager manager = EasyMock.createNiceMock(Manager.class);
- expect(manager.getSteadyTime()).andReturn(SteadyTime.from(100000, TimeUnit.NANOSECONDS))
- .anyTimes();
-
- EasyMock.replay(context, creds, security, manager);
-
- var coordinator = new TestCoordinator(context, security, new ArrayList<>(), manager);
- TNextCompactionJob nextJob = coordinator.getCompactionJob(TraceUtil.traceInfo(), creds,
++ var coordinator = new TestCoordinator(manager, new ArrayList<>());
++ TNextCompactionJob nextJob = coordinator.getCompactionJob(TraceUtil.traceInfo(), rpcCreds,
+ GROUP_ID.toString(), "localhost:10240", UUID.randomUUID().toString());
+ assertEquals(3, nextJob.getCompactorCount());
+ assertNull(nextJob.getJob().getExternalCompactionId());
-
- EasyMock.verify(context, creds, security);
+ }
+
+ @Test
+ public void testCleanUpRunning() throws Exception {
-
- ServerContext context = EasyMock.createNiceMock(ServerContext.class);
- expect(context.getCaches()).andReturn(Caches.getInstance()).anyTimes();
- expect(context.getConfiguration()).andReturn(DefaultConfiguration.getInstance()).anyTimes();
-
- TCredentials creds = EasyMock.createNiceMock(TCredentials.class);
-
- AuditedSecurityOperation security = EasyMock.createNiceMock(AuditedSecurityOperation.class);
- Manager manager = EasyMock.createNiceMock(Manager.class);
- expect(manager.getSteadyTime()).andReturn(SteadyTime.from(100000, TimeUnit.NANOSECONDS))
- .anyTimes();
-
- TExternalCompaction ext1 = EasyMock.createMock(TExternalCompaction.class);
- expect(ext1.getJob()).andReturn(new TExternalCompactionJob()).anyTimes();
- expect(ext1.getCompactor()).andReturn("localhost:9133").anyTimes();
- expect(ext1.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).anyTimes();
- TExternalCompaction ext2 = EasyMock.createMock(TExternalCompaction.class);
- expect(ext2.getJob()).andReturn(new TExternalCompactionJob()).anyTimes();
- expect(ext2.getCompactor()).andReturn("localhost:9133").anyTimes();
- expect(ext2.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).anyTimes();
- TExternalCompaction ext3 = EasyMock.createMock(TExternalCompaction.class);
- expect(ext3.getJob()).andReturn(new TExternalCompactionJob()).anyTimes();
- expect(ext3.getCompactor()).andReturn("localhost:9133").anyTimes();
- expect(ext3.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).anyTimes();
-
- EasyMock.replay(context, creds, security, manager, ext1, ext2, ext3);
-
- TestCoordinator coordinator =
- new TestCoordinator(context, security, new ArrayList<>(), manager);
++ TExternalCompaction ext1 = createMock(TExternalCompaction.class);
++ expect(ext1.getJob()).andReturn(new TExternalCompactionJob()).atLeastOnce();
++ expect(ext1.getCompactor()).andReturn("localhost:9133").atLeastOnce();
++ expect(ext1.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce();
++ TExternalCompaction ext2 = createMock(TExternalCompaction.class);
++ expect(ext2.getJob()).andReturn(new TExternalCompactionJob()).atLeastOnce();
++ expect(ext2.getCompactor()).andReturn("localhost:9133").atLeastOnce();
++ expect(ext2.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce();
++ TExternalCompaction ext3 = createMock(TExternalCompaction.class);
++ expect(ext3.getJob()).andReturn(new TExternalCompactionJob()).atLeastOnce();
++ expect(ext3.getCompactor()).andReturn("localhost:9133").atLeastOnce();
++ expect(ext3.getGroupName()).andReturn(Constants.DEFAULT_RESOURCE_GROUP_NAME).atLeastOnce();
++ replay(ext1, ext2, ext3);
++
++ TestCoordinator coordinator = new TestCoordinator(manager, new ArrayList<>());
+
+ var ecid1 = ExternalCompactionId.generate(UUID.randomUUID());
+ var ecid2 = ExternalCompactionId.generate(UUID.randomUUID());
+ var ecid3 = ExternalCompactionId.generate(UUID.randomUUID());
+
+ coordinator.getRunning().put(ecid1, new RunningCompaction(ext1));
+ coordinator.getRunning().put(ecid2, new RunningCompaction(ext2));
+ coordinator.getRunning().put(ecid3, new RunningCompaction(ext3));
-
+ coordinator.cleanUpInternalState();
+
+ assertEquals(Set.of(ecid1, ecid2, ecid3), coordinator.getRunning().keySet());
+
+ coordinator.setMetadataCompactionIds(Set.of(ecid1, ecid2));
-
+ coordinator.cleanUpInternalState();
+
+ assertEquals(Set.of(ecid1, ecid2), coordinator.getRunning().keySet());
+
- EasyMock.verify(context, creds, security, manager, ext1, ext2, ext3);
-
++ verify(ext1, ext2, ext3);
+ }
+}
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
index 6afc84b4db,1188bbdf1d..7cc263e9bd
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java
@@@ -103,10 -110,11 +103,10 @@@ import org.apache.accumulo.server.data.
import org.apache.accumulo.server.fs.TooManyFilesException;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.rpc.TServerUtils;
- import org.apache.accumulo.server.security.SecurityOperation;
+ import org.apache.accumulo.server.security.AuditedSecurityOperation;
-import org.apache.accumulo.server.zookeeper.TransactionWatcher;
-import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
+import org.apache.accumulo.server.tablets.ConditionCheckerContext;
+import org.apache.accumulo.server.tablets.ConditionCheckerContext.ConditionChecker;
import org.apache.accumulo.tserver.RowLocks.RowLock;
-import org.apache.accumulo.tserver.data.ServerConditionalMutation;
import org.apache.accumulo.tserver.session.ConditionalSession;
import org.apache.accumulo.tserver.session.SummarySession;
import org.apache.accumulo.tserver.session.UpdateSession;
@@@ -132,9 -140,11 +132,9 @@@ public class TabletClientHandler implem
private static final Logger log = LoggerFactory.getLogger(TabletClientHandler.class);
private final long MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS;
- private static final long RECENTLY_SPLIT_MILLIES = MINUTES.toMillis(1);
private final TabletServer server;
- protected final TransactionWatcher watcher;
protected final ServerContext context;
- protected final SecurityOperation security;
+ protected final AuditedSecurityOperation security;
private final WriteTracker writeTracker;
private final RowLocks rowLocks = new RowLocks();
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftScanClientHandler.java
index 83ca289fa2,b254fd3e40..29f8e785fa
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftScanClientHandler.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftScanClientHandler.java
@@@ -18,7 -18,9 +18,9 @@@
*/
package org.apache.accumulo.tserver;
+import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+ import static java.util.stream.Collectors.toList;
+ import static java.util.stream.Collectors.toMap;
-import static org.apache.accumulo.core.util.UtilWaitThread.sleepUninterruptibly;
import java.io.IOException;
import java.nio.ByteBuffer;
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TservConstraintEnv.java
index d904a2e74c,f50f9264a5..dd84bb8315
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TservConstraintEnv.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TservConstraintEnv.java
@@@ -18,20 -18,21 +18,20 @@@
*/
package org.apache.accumulo.tserver;
+ import static java.util.Collections.singletonList;
+
import java.nio.ByteBuffer;
- import java.util.Collections;
import org.apache.accumulo.core.data.TabletId;
+import org.apache.accumulo.core.data.constraints.Constraint;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.dataImpl.TabletIdImpl;
import org.apache.accumulo.core.security.AuthorizationContainer;
import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.constraints.SystemEnvironment;
- import org.apache.accumulo.server.security.SecurityOperation;
-@SuppressWarnings("deprecation")
-public class TservConstraintEnv
- implements SystemEnvironment, org.apache.accumulo.core.constraints.Constraint.Environment {
+public class TservConstraintEnv implements SystemEnvironment, Constraint.Environment {
private final ServerContext context;
private final TCredentials credentials;