You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2025/09/02 19:00:31 UTC
(accumulo) branch main updated: Added resource group parameter to client Thrift connection methods (#5775)
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new 2afd5ce3e3 Added resource group parameter to client Thrift connection methods (#5775)
2afd5ce3e3 is described below
commit 2afd5ce3e30e08482d79a7ea2f0aedac6fb322e3
Author: Dave Marion <dl...@apache.org>
AuthorDate: Tue Sep 2 15:00:25 2025 -0400
Added resource group parameter to client Thrift connection methods (#5775)
The new resource group parameter has been added in anticipation
of the resource operations api addition in #5749. This change
also adds a debug system property like we have for hosts that
allows the user to configure the client to only connect to
hosts in the resource group for non-management api operations.
Co-authored-by: Christopher Tubbs <ct...@apache.org>
---
.../accumulo/core/clientImpl/ClientContext.java | 8 +-
.../core/clientImpl/ClientTabletCacheImpl.java | 9 +-
.../core/clientImpl/InstanceOperationsImpl.java | 43 ++++---
.../core/clientImpl/NamespaceOperationsImpl.java | 22 ++--
.../core/clientImpl/SecurityOperationsImpl.java | 7 +-
.../core/clientImpl/TableOperationsImpl.java | 41 ++++---
.../core/clientImpl/ThriftTransportPool.java | 55 ++++++++-
.../core/clientImpl/ZookeeperLockChecker.java | 9 +-
.../accumulo/core/lock/ServiceLockPaths.java | 17 ++-
.../core/metadata/schema/TabletMetadata.java | 3 +-
.../rpc/clients/ClientServiceThriftClient.java | 13 ++-
.../core/rpc/clients/ManagerThriftClient.java | 5 +-
.../accumulo/core/rpc/clients/TServerClient.java | 103 ++++++++++++-----
.../TabletManagementClientServiceThriftClient.java | 13 ++-
.../core/rpc/clients/TabletServerThriftClient.java | 13 ++-
.../core/rpc/clients/ThriftClientTypes.java | 5 +-
.../org/apache/accumulo/core/summary/Gatherer.java | 3 +-
.../util/compaction/ExternalCompactionUtil.java | 33 +++---
.../accumulo/core/lock/ServiceLockPathsTest.java | 126 +++++++++++----------
.../miniclusterImpl/MiniAccumuloClusterImpl.java | 16 +--
.../accumulo/server/manager/LiveTServerSet.java | 6 +-
.../server/manager/state/DeadServerList.java | 5 +-
.../accumulo/server/util/AccumuloStatus.java | 5 +-
.../org/apache/accumulo/server/util/Admin.java | 25 ++--
.../accumulo/server/util/ServiceStatusCmd.java | 13 ++-
.../accumulo/server/util/TabletServerLocks.java | 9 +-
.../org/apache/accumulo/server/util/ZooZap.java | 2 +-
.../java/org/apache/accumulo/manager/Manager.java | 5 +-
.../accumulo/shell/commands/ListBulkCommand.java | 4 +-
.../accumulo/shell/commands/ListScansCommand.java | 3 +-
.../accumulo/test/DetectDeadTabletServersIT.java | 4 +-
.../org/apache/accumulo/test/GetManagerStats.java | 4 +-
.../java/org/apache/accumulo/test/RecoveryIT.java | 13 ++-
.../test/ScanServerConcurrentTabletScanIT.java | 3 +-
.../test/ScanServerGroupConfigurationIT.java | 33 +++---
.../org/apache/accumulo/test/ScanServerIT.java | 3 +-
.../accumulo/test/ScanServerMetadataEntriesIT.java | 3 +-
.../accumulo/test/ScanServerMultipleScansIT.java | 3 +-
.../apache/accumulo/test/ScanServerShutdownIT.java | 5 +-
.../apache/accumulo/test/TransportCachingIT.java | 34 ++++--
.../CompactionPriorityQueueMetricsIT.java | 7 +-
.../test/conf/PropStoreConfigIT_SimpleSuite.java | 8 +-
.../accumulo/test/fate/FateOpsCommandsITBase.java | 7 +-
.../functional/BalanceAfterCommsFailureIT.java | 4 +-
.../BalanceInPresenceOfOfflineTableIT.java | 4 +-
.../test/functional/DebugClientConnectionIT.java | 56 +++++++++
.../test/functional/GracefulShutdownIT.java | 7 +-
.../accumulo/test/functional/ManagerApiIT.java | 9 +-
.../test/functional/ManagerAssignmentIT.java | 10 +-
.../test/functional/MemoryStarvedMajCIT.java | 6 +-
.../test/functional/MetadataMaxFilesIT.java | 4 +-
.../test/functional/SimpleBalancerFairnessIT.java | 7 +-
.../functional/TabletManagementIteratorIT.java | 5 +-
.../functional/TabletResourceGroupBalanceIT.java | 3 +-
.../accumulo/test/lock/ServiceLockPathsIT.java | 104 ++++++++---------
.../accumulo/test/manager/SuspendedTabletsIT.java | 3 +-
.../apache/accumulo/test/upgrade/UpgradeIT.java | 13 ++-
57 files changed, 632 insertions(+), 351 deletions(-)
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
index dc5a03288e..f55baba9b8 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java
@@ -94,6 +94,7 @@ import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.manager.state.tables.TableState;
import org.apache.accumulo.core.metadata.MetadataCachedTabletObtainer;
@@ -218,8 +219,9 @@ public class ClientContext implements AccumuloClient {
@Override
public Supplier<Collection<ScanServerInfo>> getScanServers() {
- return () -> getServerPaths().getScanServer(rg -> true, AddressSelector.all(), true)
- .stream().map(entry -> new ScanServerInfo() {
+ return () -> getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).stream()
+ .map(entry -> new ScanServerInfo() {
@Override
public String getAddress() {
return entry.getServer();
@@ -455,7 +457,7 @@ public class ClientContext implements AccumuloClient {
ensureOpen();
Map<String,Pair<UUID,ResourceGroupId>> liveScanServers = new HashMap<>();
Set<ServiceLockPath> scanServerPaths =
- getServerPaths().getScanServer(rg -> true, AddressSelector.all(), true);
+ getServerPaths().getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
for (ServiceLockPath path : scanServerPaths) {
try {
ZcStat stat = new ZcStat();
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImpl.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImpl.java
index 8e4e89b435..90bd2c0ce1 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientTabletCacheImpl.java
@@ -51,6 +51,7 @@ import org.apache.accumulo.core.data.Range;
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.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.state.tables.TableState;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
@@ -675,9 +676,11 @@ public class ClientTabletCacheImpl extends ClientTabletCache {
if (!extentsToBringOnline.isEmpty()) {
log.debug("Requesting hosting for {} ondemand tablets for table id {}.",
extentsToBringOnline.size(), tableId);
- ThriftClientTypes.MANAGER.executeVoid(context,
- client -> client.requestTabletHosting(TraceUtil.traceInfo(), context.rpcCreds(),
- tableId.canonical(), extentsToBringOnline));
+ ThriftClientTypes.MANAGER
+ .executeVoid(context,
+ client -> client.requestTabletHosting(TraceUtil.traceInfo(), context.rpcCreds(),
+ tableId.canonical(), extentsToBringOnline),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
tabletHostingRequestCount.addAndGet(extentsToBringOnline.size());
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java
index bd1bd96b13..5c03648163 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java
@@ -106,7 +106,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
+ " setting its replacement {} instead", property, replacement);
});
ThriftClientTypes.MANAGER.executeVoid(context, client -> client
- .setSystemProperty(TraceUtil.traceInfo(), context.rpcCreds(), property, value));
+ .setSystemProperty(TraceUtil.traceInfo(), context.rpcCreds(), property, value),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
checkLocalityGroups(property);
}
@@ -115,7 +116,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
checkArgument(mapMutator != null, "mapMutator is null");
final TVersionedProperties vProperties = ThriftClientTypes.CLIENT.execute(context,
- client -> client.getVersionedSystemProperties(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getVersionedSystemProperties(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.ANY);
mapMutator.accept(vProperties.getProperties());
// A reference to the map was passed to the user, maybe they still have the reference and are
@@ -139,7 +141,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
// Send to server
ThriftClientTypes.MANAGER.executeVoid(context, client -> client
- .modifySystemProperties(TraceUtil.traceInfo(), context.rpcCreds(), vProperties));
+ .modifySystemProperties(TraceUtil.traceInfo(), context.rpcCreds(), vProperties),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
return vProperties.getProperties();
}
@@ -185,7 +188,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
+ " its replacement {} and will remove that instead", property, replacement);
});
ThriftClientTypes.MANAGER.executeVoid(context,
- client -> client.removeSystemProperty(TraceUtil.traceInfo(), context.rpcCreds(), property));
+ client -> client.removeSystemProperty(TraceUtil.traceInfo(), context.rpcCreds(), property),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
checkLocalityGroups(property);
}
@@ -208,21 +212,24 @@ public class InstanceOperationsImpl implements InstanceOperations {
public Map<String,String> getSystemConfiguration()
throws AccumuloException, AccumuloSecurityException {
return ThriftClientTypes.CLIENT.execute(context, client -> client
- .getConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), ConfigurationType.CURRENT));
+ .getConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), ConfigurationType.CURRENT),
+ ResourceGroupPredicate.ANY);
}
@Override
public Map<String,String> getSiteConfiguration()
throws AccumuloException, AccumuloSecurityException {
return ThriftClientTypes.CLIENT.execute(context, client -> client
- .getConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), ConfigurationType.SITE));
+ .getConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), ConfigurationType.SITE),
+ ResourceGroupPredicate.ANY);
}
@Override
public Map<String,String> getSystemProperties()
throws AccumuloException, AccumuloSecurityException {
return ThriftClientTypes.CLIENT.execute(context,
- client -> client.getSystemProperties(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getSystemProperties(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.ANY);
}
@Override
@@ -241,7 +248,7 @@ public class InstanceOperationsImpl implements InstanceOperations {
@Deprecated(since = "4.0.0")
public Set<String> getCompactors() {
Set<String> results = new HashSet<>();
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.all(), true)
+ context.getServerPaths().getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true)
.forEach(t -> results.add(t.getServer()));
return results;
}
@@ -250,7 +257,7 @@ public class InstanceOperationsImpl implements InstanceOperations {
@Deprecated(since = "4.0.0")
public Set<String> getScanServers() {
Set<String> results = new HashSet<>();
- context.getServerPaths().getScanServer(rg -> true, AddressSelector.all(), true)
+ context.getServerPaths().getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true)
.forEach(t -> results.add(t.getServer()));
return results;
}
@@ -259,7 +266,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
@Deprecated(since = "4.0.0")
public List<String> getTabletServers() {
List<String> results = new ArrayList<>();
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.all(), true)
+ context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true)
.forEach(t -> results.add(t.getServer()));
return results;
}
@@ -323,7 +331,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
public boolean testClassLoad(final String className, final String asTypeName)
throws AccumuloException, AccumuloSecurityException {
return ThriftClientTypes.CLIENT.execute(context, client -> client
- .checkClass(TraceUtil.traceInfo(), context.rpcCreds(), className, asTypeName));
+ .checkClass(TraceUtil.traceInfo(), context.rpcCreds(), className, asTypeName),
+ ResourceGroupPredicate.ANY);
}
@Override
@@ -457,7 +466,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
public void waitForBalance() throws AccumuloException {
try {
ThriftClientTypes.MANAGER.executeVoid(context,
- client -> client.waitForBalance(TraceUtil.traceInfo()));
+ client -> client.waitForBalance(TraceUtil.traceInfo()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
} catch (AccumuloSecurityException ex) {
// should never happen
throw new IllegalStateException("Unexpected exception thrown", ex);
@@ -473,7 +483,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
@Override
public Duration getManagerTime() throws AccumuloException, AccumuloSecurityException {
return Duration.ofNanos(ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerTimeNanos(TraceUtil.traceInfo(), context.rpcCreds())));
+ client -> client.getManagerTimeNanos(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY));
}
@Override
@@ -482,8 +493,8 @@ public class InstanceOperationsImpl implements InstanceOperations {
Objects.requireNonNull(type, "type parameter cannot be null");
Objects.requireNonNull(host, "host parameter cannot be null");
- final ResourceGroupPredicate rg =
- resourceGroup == null ? rgt -> true : rgt -> rgt.equals(resourceGroup);
+ final ResourceGroupPredicate rg = resourceGroup == null ? ResourceGroupPredicate.ANY
+ : ResourceGroupPredicate.exact(resourceGroup);
final AddressSelector hp = AddressSelector.exact(HostAndPort.fromParts(host, port));
switch (type) {
@@ -530,7 +541,7 @@ public class InstanceOperationsImpl implements InstanceOperations {
@Override
public Set<ServerId> getServers(ServerId.Type type) {
- return getServers(type, rg -> true, AddressSelector.all());
+ return getServers(type, ResourceGroupPredicate.ANY, AddressSelector.all());
}
@Override
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceOperationsImpl.java
index 1792208687..0e12b76792 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/NamespaceOperationsImpl.java
@@ -53,6 +53,7 @@ import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.constraints.Constraint;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.TFateOperation;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.trace.TraceUtil;
@@ -206,9 +207,10 @@ public class NamespaceOperationsImpl extends NamespaceOperationsHelper {
final Consumer<Map<String,String>> mapMutator)
throws AccumuloException, AccumuloSecurityException, NamespaceNotFoundException {
- final TVersionedProperties vProperties =
- ThriftClientTypes.CLIENT.execute(context, client -> client
- .getVersionedNamespaceProperties(TraceUtil.traceInfo(), context.rpcCreds(), namespace));
+ final TVersionedProperties vProperties = ThriftClientTypes.CLIENT.execute(context,
+ client -> client.getVersionedNamespaceProperties(TraceUtil.traceInfo(), context.rpcCreds(),
+ namespace),
+ ResourceGroupPredicate.ANY);
mapMutator.accept(vProperties.getProperties());
// A reference to the map was passed to the user, maybe they still have the reference and are
@@ -293,7 +295,8 @@ public class NamespaceOperationsImpl extends NamespaceOperationsHelper {
try {
return ThriftClientTypes.CLIENT.execute(context, client -> client
- .getNamespaceConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), namespace));
+ .getNamespaceConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), namespace),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloSecurityException e) {
throw e;
} catch (AccumuloException e) {
@@ -319,7 +322,8 @@ public class NamespaceOperationsImpl extends NamespaceOperationsHelper {
try {
return ThriftClientTypes.CLIENT.execute(context, client -> client
- .getNamespaceProperties(TraceUtil.traceInfo(), context.rpcCreds(), namespace));
+ .getNamespaceProperties(TraceUtil.traceInfo(), context.rpcCreds(), namespace),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloException e) {
Throwable eCause = e.getCause();
if (eCause instanceof TableNotFoundException) {
@@ -353,9 +357,11 @@ public class NamespaceOperationsImpl extends NamespaceOperationsHelper {
checkArgument(asTypeName != null, "asTypeName is null");
try {
- return ThriftClientTypes.CLIENT.execute(context,
- client -> client.checkNamespaceClass(TraceUtil.traceInfo(), context.rpcCreds(), namespace,
- className, asTypeName));
+ return ThriftClientTypes.CLIENT
+ .execute(
+ context, client -> client.checkNamespaceClass(TraceUtil.traceInfo(),
+ context.rpcCreds(), namespace, className, asTypeName),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloSecurityException e) {
throw e;
} catch (AccumuloException e) {
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/SecurityOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/SecurityOperationsImpl.java
index d074b1544e..b053bcda2d 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/SecurityOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/SecurityOperationsImpl.java
@@ -38,6 +38,7 @@ import org.apache.accumulo.core.clientImpl.thrift.ClientService;
import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType;
import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes.Exec;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes.ExecVoid;
@@ -64,7 +65,8 @@ public class SecurityOperationsImpl implements SecurityOperations {
private void executeVoid(ExecVoid<ClientService.Client> exec)
throws AccumuloException, AccumuloSecurityException {
try {
- ThriftClientTypes.CLIENT.executeVoid(context, client -> exec.execute(client));
+ ThriftClientTypes.CLIENT.executeVoid(context, client -> exec.execute(client),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloSecurityException e) {
throw e;
} catch (AccumuloException e) {
@@ -101,7 +103,8 @@ public class SecurityOperationsImpl implements SecurityOperations {
private <R> R execute(Exec<R,ClientService.Client> exec)
throws AccumuloException, AccumuloSecurityException {
try {
- return ThriftClientTypes.CLIENT.execute(context, client -> exec.execute(client));
+ return ThriftClientTypes.CLIENT.execute(context, client -> exec.execute(client),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloSecurityException e) {
throw e;
} catch (AccumuloException e) {
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java
index 95b81a86cf..5a19fab106 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/TableOperationsImpl.java
@@ -143,6 +143,7 @@ import org.apache.accumulo.core.dataImpl.thrift.TSummaryRequest;
import org.apache.accumulo.core.fate.FateInstanceType;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.state.tables.TableState;
import org.apache.accumulo.core.manager.thrift.FateService;
import org.apache.accumulo.core.manager.thrift.ManagerClientService;
@@ -1053,9 +1054,10 @@ public class TableOperationsImpl extends TableOperationsHelper {
private Map<String,String> tryToModifyProperties(String tableName,
final Consumer<Map<String,String>> mapMutator) throws AccumuloException,
AccumuloSecurityException, IllegalArgumentException, ConcurrentModificationException {
- final TVersionedProperties vProperties =
- ThriftClientTypes.CLIENT.execute(context, client -> client
- .getVersionedTableProperties(TraceUtil.traceInfo(), context.rpcCreds(), tableName));
+ final TVersionedProperties vProperties = ThriftClientTypes.CLIENT.execute(context,
+ client -> client.getVersionedTableProperties(TraceUtil.traceInfo(), context.rpcCreds(),
+ tableName),
+ ResourceGroupPredicate.ANY);
mapMutator.accept(vProperties.getProperties());
// A reference to the map was passed to the user, maybe they still have the reference and are
@@ -1066,9 +1068,11 @@ public class TableOperationsImpl extends TableOperationsHelper {
try {
// Send to server
- ThriftClientTypes.MANAGER.executeVoid(context,
- client -> client.modifyTableProperties(TraceUtil.traceInfo(), context.rpcCreds(),
- tableName, vProperties));
+ ThriftClientTypes.MANAGER
+ .executeVoid(
+ context, client -> client.modifyTableProperties(TraceUtil.traceInfo(),
+ context.rpcCreds(), tableName, vProperties),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
for (String property : vProperties.getProperties().keySet()) {
checkLocalityGroups(tableName, property);
}
@@ -1135,7 +1139,8 @@ public class TableOperationsImpl extends TableOperationsHelper {
final String value)
throws AccumuloException, AccumuloSecurityException, TableNotFoundException {
ThriftClientTypes.MANAGER.executeVoid(context, client -> client
- .setTableProperty(TraceUtil.traceInfo(), context.rpcCreds(), tableName, property, value));
+ .setTableProperty(TraceUtil.traceInfo(), context.rpcCreds(), tableName, property, value),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
@Override
@@ -1156,7 +1161,8 @@ public class TableOperationsImpl extends TableOperationsHelper {
private void removePropertyNoChecks(final String tableName, final String property)
throws AccumuloException, AccumuloSecurityException, TableNotFoundException {
ThriftClientTypes.MANAGER.executeVoid(context, client -> client
- .removeTableProperty(TraceUtil.traceInfo(), context.rpcCreds(), tableName, property));
+ .removeTableProperty(TraceUtil.traceInfo(), context.rpcCreds(), tableName, property),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
void checkLocalityGroups(String tableName, String propChanged)
@@ -1180,7 +1186,8 @@ public class TableOperationsImpl extends TableOperationsHelper {
try {
return ThriftClientTypes.CLIENT.execute(context, client -> client
- .getTableConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), tableName));
+ .getTableConfiguration(TraceUtil.traceInfo(), context.rpcCreds(), tableName),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloException e) {
Throwable t = e.getCause();
if (t instanceof TableNotFoundException) {
@@ -1200,8 +1207,9 @@ public class TableOperationsImpl extends TableOperationsHelper {
EXISTING_TABLE_NAME.validate(tableName);
try {
- return ThriftClientTypes.CLIENT.execute(context, client -> client
- .getTableProperties(TraceUtil.traceInfo(), context.rpcCreds(), tableName));
+ return ThriftClientTypes.CLIENT.execute(context,
+ client -> client.getTableProperties(TraceUtil.traceInfo(), context.rpcCreds(), tableName),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloException e) {
Throwable t = e.getCause();
if (t instanceof TableNotFoundException) {
@@ -1589,7 +1597,8 @@ public class TableOperationsImpl extends TableOperationsHelper {
// this operation may us a lot of memory... it's likely that connections to tabletservers
// hosting metadata tablets will be cached, so do not use cached
// connections
- pair = ThriftClientTypes.CLIENT.getThriftServerConnection(context, false);
+ pair = ThriftClientTypes.CLIENT.getThriftServerConnection(context, false,
+ ResourceGroupPredicate.ANY);
diskUsages = pair.getSecond().getDiskUsage(tableNames, context.rpcCreds());
} catch (ThriftTableOperationException e) {
switch (e.getType()) {
@@ -1787,7 +1796,8 @@ public class TableOperationsImpl extends TableOperationsHelper {
try {
return ThriftClientTypes.CLIENT.execute(context,
client -> client.checkTableClass(TraceUtil.traceInfo(), context.rpcCreds(), tableName,
- className, asTypeName));
+ className, asTypeName),
+ ResourceGroupPredicate.ANY);
} catch (AccumuloSecurityException e) {
throw e;
} catch (AccumuloException e) {
@@ -2080,7 +2090,7 @@ public class TableOperationsImpl extends TableOperationsHelper {
tsr = client.contiuneGetSummaries(TraceUtil.traceInfo(), tsr.sessionId);
}
return tsr;
- });
+ }, ResourceGroupPredicate.ANY);
return new SummaryCollection(ret).getSummaries();
}
@@ -2289,7 +2299,8 @@ public class TableOperationsImpl extends TableOperationsHelper {
var currentTime = Suppliers.memoize(() -> {
try {
return Duration.ofNanos(ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerTimeNanos(TraceUtil.traceInfo(), context.rpcCreds())));
+ client -> client.getManagerTimeNanos(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY));
} catch (AccumuloException | AccumuloSecurityException e) {
throw new IllegalStateException(e);
}
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java
index 5090cfa516..e9dea2705b 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java
@@ -35,16 +35,24 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
+import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
+import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.util.Timer;
import org.apache.accumulo.core.util.threads.Threads;
+import org.apache.commons.collections4.set.CompositeSet;
import org.apache.thrift.TConfiguration;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
@@ -137,10 +145,50 @@ public class ThriftTransportPool {
return createNewTransport(cacheKey);
}
- public Pair<String,TTransport> getAnyCachedTransport(ThriftClientTypes<?> type) {
+ @SuppressWarnings("unchecked")
+ public Pair<String,TTransport> getAnyCachedTransport(ThriftClientTypes<?> type, ClientContext ctx,
+ ThriftService service, ResourceGroupPredicate rgp) {
+
+ final Timer timer = Timer.startNew();
final List<ThriftTransportKey> serversSet = new ArrayList<>();
+
+ Function<AddressSelector,Set<ServiceLockPath>> paths = null;
+ switch (service) {
+ case CLIENT:
+ paths = (selector) -> new CompositeSet<ServiceLockPath>(
+ ctx.getServerPaths().getCompactor(rgp, selector, true),
+ ctx.getServerPaths().getScanServer(rgp, selector, true),
+ ctx.getServerPaths().getTabletServer(rgp, selector, true));
+ break;
+ case COMPACTOR:
+ paths = (selector) -> ctx.getServerPaths().getCompactor(rgp, selector, true);
+ break;
+ case MANAGER:
+ case COORDINATOR:
+ case FATE:
+ paths = (selector) -> Set.of(ctx.getServerPaths().getManager(true));
+ break;
+ case GC:
+ paths = (selector) -> Set.of(ctx.getServerPaths().getGarbageCollector(true));
+ break;
+ case TABLET_SCAN:
+ paths = (selector) -> new CompositeSet<ServiceLockPath>(
+ ctx.getServerPaths().getTabletServer(rgp, selector, true),
+ ctx.getServerPaths().getScanServer(rgp, selector, true));
+ break;
+ case TABLET_INGEST:
+ case TABLET_MANAGEMENT:
+ case TSERV:
+ paths = (selector) -> ctx.getServerPaths().getTabletServer(rgp, selector, true);
+ break;
+ case NONE:
+ default:
+ throw new IllegalArgumentException("Unhandled thrift service type: " + service);
+ }
+
for (ThriftTransportKey ttk : connectionPool.getThriftTransportKeys()) {
- if (ttk.getType().equals(type)) {
+ if (ttk.getType().equals(type)
+ && !paths.apply(AddressSelector.exact(ttk.getServer())).isEmpty()) {
serversSet.add(ttk);
}
}
@@ -152,7 +200,8 @@ public class ThriftTransportPool {
CachedConnection connection = connectionPool.reserveAny(ttk);
if (connection != null) {
final String serverAddr = ttk.getServer().toString();
- log.trace("Using existing connection to {}", serverAddr);
+ log.trace("Took {} ms to evaluate existing connection to {}",
+ timer.elapsed(TimeUnit.MILLISECONDS), serverAddr);
return new Pair<>(serverAddr, connection.transport);
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java
index d83525738e..68382a867c 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ZookeeperLockChecker.java
@@ -27,6 +27,7 @@ import org.apache.accumulo.core.clientImpl.ClientTabletCacheImpl.TabletServerLoc
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockPaths;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.zookeeper.ZooCache;
@@ -45,8 +46,8 @@ public class ZookeeperLockChecker implements TabletServerLockChecker {
public boolean doesTabletServerLockExist(String server) {
// ServiceLockPaths only returns items that have a lock
var hostAndPort = HostAndPort.fromString(server);
- Set<ServiceLockPath> tservers =
- lockPaths.getTabletServer(rg -> true, AddressSelector.exact(hostAndPort), true);
+ Set<ServiceLockPath> tservers = lockPaths.getTabletServer(ResourceGroupPredicate.ANY,
+ AddressSelector.exact(hostAndPort), true);
return !tservers.isEmpty();
}
@@ -54,8 +55,8 @@ public class ZookeeperLockChecker implements TabletServerLockChecker {
public boolean isLockHeld(String server, String session) {
// ServiceLockPaths only returns items that have a lock
var hostAndPort = HostAndPort.fromString(server);
- Set<ServiceLockPath> tservers =
- lockPaths.getTabletServer(rg -> true, AddressSelector.exact(hostAndPort), true);
+ Set<ServiceLockPath> tservers = lockPaths.getTabletServer(ResourceGroupPredicate.ANY,
+ AddressSelector.exact(hostAndPort), true);
for (ServiceLockPath slp : tservers) {
if (ServiceLock.getSessionId(zc, slp) == Long.parseLong(session, 16)) {
return true;
diff --git a/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java b/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java
index ab266636fd..177912c703 100644
--- a/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java
+++ b/core/src/main/java/org/apache/accumulo/core/lock/ServiceLockPaths.java
@@ -296,8 +296,8 @@ public class ServiceLockPaths {
* the ZooKeeper path.
*/
public ServiceLockPath getGarbageCollector(boolean withLock) {
- Set<ServiceLockPath> results =
- get(Constants.ZGC_LOCK, rg -> true, AddressSelector.all(), withLock);
+ Set<ServiceLockPath> results = get(Constants.ZGC_LOCK, ResourceGroupPredicate.DEFAULT_RG_ONLY,
+ AddressSelector.all(), withLock);
if (results.isEmpty()) {
return null;
} else {
@@ -311,8 +311,8 @@ public class ServiceLockPaths {
* InstanceOperations.getServers(ServerId.Type.MANAGER) to get the location.
*/
public ServiceLockPath getManager(boolean withLock) {
- Set<ServiceLockPath> results =
- get(Constants.ZMANAGER_LOCK, rg -> true, AddressSelector.all(), withLock);
+ Set<ServiceLockPath> results = get(Constants.ZMANAGER_LOCK,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, AddressSelector.all(), withLock);
if (results.isEmpty()) {
return null;
} else {
@@ -326,8 +326,8 @@ public class ServiceLockPaths {
* ZooKeeper path.
*/
public ServiceLockPath getMonitor(boolean withLock) {
- Set<ServiceLockPath> results =
- get(Constants.ZMONITOR_LOCK, rg -> true, AddressSelector.all(), withLock);
+ Set<ServiceLockPath> results = get(Constants.ZMONITOR_LOCK,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, AddressSelector.all(), withLock);
if (results.isEmpty()) {
return null;
} else {
@@ -351,7 +351,12 @@ public class ServiceLockPaths {
}
public interface ResourceGroupPredicate extends Predicate<ResourceGroupId> {
+ ResourceGroupPredicate ANY = rgid -> true;
+ ResourceGroupPredicate DEFAULT_RG_ONLY = ResourceGroupId.DEFAULT::equals;
+ static ResourceGroupPredicate exact(ResourceGroupId rgid) {
+ return rgid::equals;
+ }
}
public static class AddressSelector {
diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
index 5b0739a7bd..3ae83c4eca 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
@@ -71,6 +71,7 @@ import org.apache.accumulo.core.fate.FateId;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.SuspendingTServer;
@@ -829,7 +830,7 @@ public class TabletMetadata {
public static synchronized Set<TServerInstance> getLiveTServers(ClientContext context) {
final Set<TServerInstance> liveServers = new HashSet<>();
- for (ServiceLockPath slp : context.getServerPaths().getTabletServer(rg -> true,
+ for (ServiceLockPath slp : context.getServerPaths().getTabletServer(ResourceGroupPredicate.ANY,
AddressSelector.all(), true)) {
checkTabletServer(context, slp).ifPresent(liveServers::add);
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java b/core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java
index c1dc9954da..a25aee50ee 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java
@@ -25,6 +25,7 @@ import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.clientImpl.thrift.ClientService.Client;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.util.Pair;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
@@ -42,21 +43,21 @@ public class ClientServiceThriftClient extends ThriftClientTypes<Client>
@Override
public Pair<String,Client> getThriftServerConnection(ClientContext context,
- boolean preferCachedConnections) throws TTransportException {
+ boolean preferCachedConnections, ResourceGroupPredicate rgp) throws TTransportException {
return getThriftServerConnection(LOG, this, context, preferCachedConnections,
- warnedAboutTServersBeingDown, ThriftService.CLIENT);
+ warnedAboutTServersBeingDown, ThriftService.CLIENT, rgp);
}
@Override
- public <R> R execute(ClientContext context, Exec<R,Client> exec)
+ public <R> R execute(ClientContext context, Exec<R,Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
- return execute(LOG, context, exec);
+ return execute(LOG, context, exec, rgp);
}
@Override
- public void executeVoid(ClientContext context, ExecVoid<Client> exec)
+ public void executeVoid(ClientContext context, ExecVoid<Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
- executeVoid(LOG, context, exec);
+ executeVoid(LOG, context, exec, rgp);
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/clients/ManagerThriftClient.java b/core/src/main/java/org/apache/accumulo/core/rpc/clients/ManagerThriftClient.java
index eb353feb90..097aa8a734 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/clients/ManagerThriftClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/clients/ManagerThriftClient.java
@@ -32,6 +32,7 @@ import org.apache.accumulo.core.clientImpl.thrift.ThriftConcurrentModificationEx
import org.apache.accumulo.core.clientImpl.thrift.ThriftNotActiveServiceException;
import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerClientService.Client;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.thrift.transport.TTransportException;
@@ -90,7 +91,7 @@ public class ManagerThriftClient extends ThriftClientTypes<Client>
}
@Override
- public <R> R execute(ClientContext context, Exec<R,Client> exec)
+ public <R> R execute(ClientContext context, Exec<R,Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
try {
return executeTableCommand(context, exec);
@@ -138,7 +139,7 @@ public class ManagerThriftClient extends ThriftClientTypes<Client>
}
@Override
- public void executeVoid(ClientContext context, ExecVoid<Client> exec)
+ public void executeVoid(ClientContext context, ExecVoid<Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
try {
executeVoidTableCommand(context, exec);
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java
index 3539b853e6..6265d981bb 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TServerClient.java
@@ -39,9 +39,12 @@ import org.apache.accumulo.core.clientImpl.AccumuloServerException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
import org.apache.accumulo.core.clientImpl.thrift.ThriftTableOperationException;
+import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
+import org.apache.accumulo.core.lock.ServiceLockPaths;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes.Exec;
@@ -60,20 +63,51 @@ import com.google.common.net.HostAndPort;
public interface TServerClient<C extends TServiceClient> {
static final String DEBUG_HOST = "org.apache.accumulo.client.rpc.debug.host";
+ static final String DEBUG_RG = "org.apache.accumulo.client.rpc.debug.group";
- Pair<String,C> getThriftServerConnection(ClientContext context, boolean preferCachedConnections)
- throws TTransportException;
+ Pair<String,C> getThriftServerConnection(ClientContext context, boolean preferCachedConnections,
+ ResourceGroupPredicate rgp) throws TTransportException;
default Pair<String,C> getThriftServerConnection(Logger LOG, ThriftClientTypes<C> type,
ClientContext context, boolean preferCachedConnections, AtomicBoolean warned,
- ThriftService service) throws TTransportException {
+ ThriftService service, ResourceGroupPredicate rgp) throws TTransportException {
checkArgument(context != null, "context is null");
final String debugHost = System.getProperty(DEBUG_HOST, null);
+ final boolean debugHostSpecified = debugHost != null;
+ final String debugRG = System.getProperty(DEBUG_RG, null);
+ final boolean debugRGSpecified = debugRG != null;
+ final ResourceGroupId debugRGid = debugRGSpecified ? ResourceGroupId.of(debugRG) : null;
- if (preferCachedConnections && debugHost == null) {
+ if (debugHostSpecified && debugRGSpecified) {
+ LOG.warn("System properties {} and {} are both set. If set incorrectly then"
+ + " this client may not find a server to connect to.", DEBUG_HOST, DEBUG_RG);
+ }
+
+ if (debugRGSpecified) {
+ if (type == ThriftClientTypes.CLIENT || type == ThriftClientTypes.COMPACTOR
+ || type == ThriftClientTypes.SERVER_PROCESS || type == ThriftClientTypes.TABLET_INGEST
+ || type == ThriftClientTypes.TABLET_MGMT || type == ThriftClientTypes.TABLET_SCAN
+ || type == ThriftClientTypes.TABLET_SERVER) {
+ if (rgp.test(debugRGid)) {
+ // its safe to potentially narrow the predicate
+ LOG.debug("System property '{}' set to '{}' overriding predicate argument", DEBUG_RG,
+ debugRG);
+ rgp = ResourceGroupPredicate.exact(debugRGid);
+ } else {
+ LOG.warn("System property '{}' set to '{}' does not intersect with predicate argument."
+ + " Ignoring degug system property.", DEBUG_RG, debugRG);
+ }
+ } else {
+ LOG.debug(
+ "System property '{}' set to '{}' but ignored when making RPCs to management servers",
+ DEBUG_RG, debugRG);
+ }
+ }
+
+ if (preferCachedConnections && !debugHostSpecified && !debugRGSpecified) {
Pair<String,TTransport> cachedTransport =
- context.getTransportPool().getAnyCachedTransport(type);
+ context.getTransportPool().getAnyCachedTransport(type, context, service, rgp);
if (cachedTransport != null) {
C client = ThriftUtil.createClient(type, cachedTransport.getSecond());
warned.set(false);
@@ -83,27 +117,23 @@ public interface TServerClient<C extends TServiceClient> {
final long rpcTimeout = context.getClientTimeoutInMillis();
final ZooCache zc = context.getZooCache();
+ final ServiceLockPaths sp = context.getServerPaths();
final List<ServiceLockPath> serverPaths = new ArrayList<>();
- if (type == ThriftClientTypes.CLIENT && debugHost != null) {
+
+ if (type == ThriftClientTypes.CLIENT && debugHostSpecified) {
// add all three paths to the set even though they may not be correct.
// The entire set will be checked in the code below to validate
// that the path is correct and the lock is held and will return the
// correct one.
HostAndPort hp = HostAndPort.fromString(debugHost);
- serverPaths.addAll(
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.exact(hp), true));
- serverPaths.addAll(
- context.getServerPaths().getScanServer(rg -> true, AddressSelector.exact(hp), true));
- serverPaths.addAll(
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.exact(hp), true));
+ serverPaths.addAll(sp.getCompactor(rgp, AddressSelector.exact(hp), true));
+ serverPaths.addAll(sp.getScanServer(rgp, AddressSelector.exact(hp), true));
+ serverPaths.addAll(sp.getTabletServer(rgp, AddressSelector.exact(hp), true));
} else {
- serverPaths.addAll(
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.all(), false));
+ serverPaths.addAll(sp.getTabletServer(rgp, AddressSelector.all(), false));
if (type == ThriftClientTypes.CLIENT) {
- serverPaths.addAll(
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.all(), false));
- serverPaths.addAll(
- context.getServerPaths().getScanServer(rg -> true, AddressSelector.all(), false));
+ serverPaths.addAll(sp.getCompactor(rgp, AddressSelector.all(), false));
+ serverPaths.addAll(sp.getScanServer(rgp, AddressSelector.all(), false));
}
if (serverPaths.isEmpty()) {
if (warned.compareAndSet(false, true)) {
@@ -111,7 +141,12 @@ public interface TServerClient<C extends TServiceClient> {
"There are no servers serving the {} api: check that zookeeper and accumulo are running.",
type);
}
- throw new TTransportException("There are no servers for type: " + type);
+ // If the user set the system property for the resource group, then don't throw
+ // a TTransportException here. That will cause the call to be continuously retried.
+ // Instead, let this continue so that we can throw a different error below.
+ if (!debugRGSpecified) {
+ throw new TTransportException("There are no servers for type: " + type);
+ }
}
}
@@ -126,16 +161,22 @@ public interface TServerClient<C extends TServiceClient> {
TTransport transport = context.getTransportPool().getTransport(type,
tserverClientAddress, rpcTimeout, context, preferCachedConnections);
C client = ThriftUtil.createClient(type, transport);
- if (type == ThriftClientTypes.CLIENT && debugHost != null) {
+ if (type == ThriftClientTypes.CLIENT && debugHostSpecified) {
LOG.info("Connecting to debug host: {}", debugHost);
}
warned.set(false);
return new Pair<String,C>(tserverClientAddress.toString(), client);
} catch (TTransportException e) {
- if (type == ThriftClientTypes.CLIENT && debugHost != null) {
+ if (type == ThriftClientTypes.CLIENT && debugHostSpecified) {
LOG.error(
- "Error creating transport to debug host: {}. If this server is down, then you will need to remove or change the system property {}.",
+ "Error creating transport to debug host: {}. If this server is"
+ + " down, then you will need to remove or change the system property {}.",
debugHost, DEBUG_HOST);
+ } else if (debugRGSpecified && rgp.test(debugRGid)) {
+ LOG.error(
+ "Error creating transport to debug group: {}. If all servers are"
+ + " down, then you will need to remove or change the system property {}.",
+ debugRG, DEBUG_RG);
} else {
LOG.trace("Error creating transport to {}", tserverClientAddress);
}
@@ -151,22 +192,26 @@ public interface TServerClient<C extends TServiceClient> {
}
// Need to throw a different exception, when a TTransportException is
// thrown below, then the operation will be retried endlessly.
- if (type == ThriftClientTypes.CLIENT && debugHost != null) {
+ if (type == ThriftClientTypes.CLIENT && debugHostSpecified) {
throw new UncheckedIOException("Error creating transport to debug host: " + debugHost
+ ". If this server is down, then you will need to remove or change the system property "
+ DEBUG_HOST + ".", new IOException(""));
+ } else if (debugRGSpecified && rgp.test(debugRGid)) {
+ throw new UncheckedIOException("Error creating transport to debug group: " + debugRG
+ + ". If all servers are down, then you will need to remove or change the system property "
+ + DEBUG_RG + ".", new IOException(""));
} else {
throw new TTransportException("Failed to connect to any server for API type " + type);
}
}
- default <R> R execute(Logger LOG, ClientContext context, Exec<R,C> exec)
- throws AccumuloException, AccumuloSecurityException {
+ default <R> R execute(Logger LOG, ClientContext context, Exec<R,C> exec,
+ ResourceGroupPredicate rgp) throws AccumuloException, AccumuloSecurityException {
while (true) {
String server = null;
C client = null;
try {
- Pair<String,C> pair = getThriftServerConnection(context, true);
+ Pair<String,C> pair = getThriftServerConnection(context, true, rgp);
server = pair.getFirst();
client = pair.getSecond();
return exec.execute(client);
@@ -200,13 +245,13 @@ public interface TServerClient<C extends TServiceClient> {
}
}
- default void executeVoid(Logger LOG, ClientContext context, ExecVoid<C> exec)
- throws AccumuloException, AccumuloSecurityException {
+ default void executeVoid(Logger LOG, ClientContext context, ExecVoid<C> exec,
+ ResourceGroupPredicate rgp) throws AccumuloException, AccumuloSecurityException {
while (true) {
String server = null;
C client = null;
try {
- Pair<String,C> pair = getThriftServerConnection(context, true);
+ Pair<String,C> pair = getThriftServerConnection(context, true, rgp);
server = pair.getFirst();
client = pair.getSecond();
exec.execute(client);
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletManagementClientServiceThriftClient.java b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletManagementClientServiceThriftClient.java
index 2b83ebf072..063002f0d7 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletManagementClientServiceThriftClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletManagementClientServiceThriftClient.java
@@ -24,6 +24,7 @@ import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.tablet.thrift.TabletManagementClientService.Client;
import org.apache.accumulo.core.util.Pair;
import org.apache.thrift.transport.TTransportException;
@@ -47,21 +48,21 @@ public class TabletManagementClientServiceThriftClient extends ThriftClientTypes
@Override
public Pair<String,Client> getThriftServerConnection(ClientContext context,
- boolean preferCachedConnections) throws TTransportException {
+ boolean preferCachedConnections, ResourceGroupPredicate rgp) throws TTransportException {
return getThriftServerConnection(LOG, this, context, preferCachedConnections,
- warnedAboutTServersBeingDown, ThriftService.TABLET_MANAGEMENT);
+ warnedAboutTServersBeingDown, ThriftService.TABLET_MANAGEMENT, rgp);
}
@Override
- public <R> R execute(ClientContext context, Exec<R,Client> exec)
+ public <R> R execute(ClientContext context, Exec<R,Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
- return execute(LOG, context, exec);
+ return execute(LOG, context, exec, rgp);
}
@Override
- public void executeVoid(ClientContext context, ExecVoid<Client> exec)
+ public void executeVoid(ClientContext context, ExecVoid<Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
- executeVoid(LOG, context, exec);
+ executeVoid(LOG, context, exec, rgp);
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletServerThriftClient.java b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletServerThriftClient.java
index 3f2f71bc7a..5cd489651c 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletServerThriftClient.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletServerThriftClient.java
@@ -24,6 +24,7 @@ import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService.Client;
import org.apache.accumulo.core.util.Pair;
import org.apache.thrift.transport.TTransportException;
@@ -46,21 +47,21 @@ public class TabletServerThriftClient extends ThriftClientTypes<Client>
@Override
public Pair<String,Client> getThriftServerConnection(ClientContext context,
- boolean preferCachedConnections) throws TTransportException {
+ boolean preferCachedConnections, ResourceGroupPredicate rgp) throws TTransportException {
return getThriftServerConnection(LOG, this, context, preferCachedConnections,
- warnedAboutTServersBeingDown, ThriftService.TSERV);
+ warnedAboutTServersBeingDown, ThriftService.TSERV, rgp);
}
@Override
- public <R> R execute(ClientContext context, Exec<R,Client> exec)
+ public <R> R execute(ClientContext context, Exec<R,Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
- return execute(LOG, context, exec);
+ return execute(LOG, context, exec, rgp);
}
@Override
- public void executeVoid(ClientContext context, ExecVoid<Client> exec)
+ public void executeVoid(ClientContext context, ExecVoid<Client> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
- executeVoid(LOG, context, exec);
+ executeVoid(LOG, context, exec, rgp);
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java b/core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java
index 5b9a5c203d..6856afef0c 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java
@@ -24,6 +24,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.thrift.TException;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.TServiceClientFactory;
@@ -115,12 +116,12 @@ public abstract class ThriftClientTypes<C extends TServiceClient> {
}
}
- public <R> R execute(ClientContext context, Exec<R,C> exec)
+ public <R> R execute(ClientContext context, Exec<R,C> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
throw new UnsupportedOperationException("This method has not been implemented");
}
- public void executeVoid(ClientContext context, ExecVoid<C> exec)
+ public void executeVoid(ClientContext context, ExecVoid<C> exec, ResourceGroupPredicate rgp)
throws AccumuloException, AccumuloSecurityException {
throw new UnsupportedOperationException("This method has not been implemented");
}
diff --git a/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java b/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java
index d9258a1130..ae60b00dc2 100644
--- a/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java
+++ b/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java
@@ -62,6 +62,7 @@ import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.dataImpl.thrift.TRowRange;
import org.apache.accumulo.core.dataImpl.thrift.TSummaries;
import org.apache.accumulo.core.dataImpl.thrift.TSummaryRequest;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
@@ -482,7 +483,7 @@ public class Gatherer {
tsr = client.contiuneGetSummaries(tinfo, tsr.sessionId);
}
return tsr;
- });
+ }, ResourceGroupPredicate.ANY);
} catch (AccumuloException | AccumuloSecurityException e) {
throw new IllegalStateException(e);
}
diff --git a/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java b/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java
index 6e450fb1b8..7b996d030e 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java
@@ -41,6 +41,7 @@ import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
import org.apache.accumulo.core.rpc.ThriftUtil;
@@ -116,10 +117,12 @@ public class ExternalCompactionUtil {
*/
public static Map<String,Set<HostAndPort>> getCompactorAddrs(ClientContext context) {
final Map<String,Set<HostAndPort>> groupsAndAddresses = new HashMap<>();
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.all(), true).forEach(slp -> {
- groupsAndAddresses.computeIfAbsent(slp.getResourceGroup().canonical(), (k) -> new HashSet<>())
- .add(HostAndPort.fromString(slp.getServer()));
- });
+ context.getServerPaths().getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true)
+ .forEach(slp -> {
+ groupsAndAddresses
+ .computeIfAbsent(slp.getResourceGroup().canonical(), (k) -> new HashSet<>())
+ .add(HostAndPort.fromString(slp.getServer()));
+ });
return groupsAndAddresses;
}
@@ -203,11 +206,12 @@ public class ExternalCompactionUtil {
final ExecutorService executor = ThreadPools.getServerThreadPools()
.getPoolBuilder(COMPACTOR_RUNNING_COMPACTIONS_POOL).numCoreThreads(16).build();
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.all(), true).forEach(slp -> {
- final HostAndPort hp = HostAndPort.fromString(slp.getServer());
- rcFutures.add(new RunningCompactionFuture(slp,
- executor.submit(() -> getRunningCompaction(hp, context))));
- });
+ context.getServerPaths().getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true)
+ .forEach(slp -> {
+ final HostAndPort hp = HostAndPort.fromString(slp.getServer());
+ rcFutures.add(new RunningCompactionFuture(slp,
+ executor.submit(() -> getRunningCompaction(hp, context))));
+ });
executor.shutdown();
final List<RunningCompaction> results = new ArrayList<>();
@@ -231,10 +235,11 @@ public class ExternalCompactionUtil {
.getPoolBuilder(COMPACTOR_RUNNING_COMPACTION_IDS_POOL).numCoreThreads(16).build();
List<Future<ExternalCompactionId>> futures = new ArrayList<>();
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.all(), true).forEach(slp -> {
- final HostAndPort hp = HostAndPort.fromString(slp.getServer());
- futures.add(executor.submit(() -> getRunningCompactionId(hp, context)));
- });
+ context.getServerPaths().getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true)
+ .forEach(slp -> {
+ final HostAndPort hp = HostAndPort.fromString(slp.getServer());
+ futures.add(executor.submit(() -> getRunningCompactionId(hp, context)));
+ });
executor.shutdown();
HashSet<ExternalCompactionId> runningIds = new HashSet<>();
@@ -256,7 +261,7 @@ public class ExternalCompactionUtil {
public static int countCompactors(ResourceGroupId group, ClientContext context) {
var start = Timer.startNew();
int count = context.getServerPaths()
- .getCompactor(rg -> rg.equals(group), AddressSelector.all(), true).size();
+ .getCompactor(ResourceGroupPredicate.exact(group), AddressSelector.all(), true).size();
long elapsed = start.elapsed(MILLISECONDS);
if (elapsed > 100) {
LOG.debug("Took {} ms to count {} compactors for {}", elapsed, count, group);
diff --git a/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java b/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java
index 7503614daa..c4043e5fb4 100644
--- a/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/lock/ServiceLockPathsTest.java
@@ -48,6 +48,7 @@ import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.zookeeper.ZcStat;
import org.apache.accumulo.core.zookeeper.ZooCache;
@@ -284,13 +285,13 @@ public class ServiceLockPathsTest {
assertThrows(NullPointerException.class, () -> paths.getCompactor(null, null, true));
assertThrows(NullPointerException.class,
- () -> paths.getCompactor(rg -> rg.equals(TEST_RESOURCE_GROUP), null, true));
- assertTrue(paths.getCompactor(rg -> true, AddressSelector.all(), true).isEmpty());
- assertTrue(paths.getCompactor(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(), true)
- .isEmpty());
+ () -> paths.getCompactor(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP), null, true));
assertTrue(
- paths.getCompactor(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.exact(hp), true)
- .isEmpty());
+ paths.getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
+ assertTrue(paths.getCompactor(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), true).isEmpty());
+ assertTrue(paths.getCompactor(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.exact(hp), true).isEmpty());
}
@Test
@@ -334,7 +335,8 @@ public class ServiceLockPathsTest {
replay(zc);
// query for all
- Set<ServiceLockPath> results = paths.getCompactor(rg -> true, AddressSelector.all(), false);
+ Set<ServiceLockPath> results =
+ paths.getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
assertEquals(4, results.size());
for (ServiceLockPath path : results) {
assertEquals(ZCOMPACTORS, path.getType());
@@ -352,7 +354,7 @@ public class ServiceLockPathsTest {
}
// query for all with locks
- results = paths.getCompactor(rg -> true, AddressSelector.all(), true);
+ results = paths.getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
assertEquals(2, results.size());
Iterator<ServiceLockPath> iter = results.iterator();
var slp1 = iter.next();
@@ -381,12 +383,14 @@ public class ServiceLockPathsTest {
}
// query for all in non-existent resource group
- results = paths.getCompactor(rg -> rg.equals(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
- AddressSelector.all(), true);
+ results =
+ paths.getCompactor(ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
+ AddressSelector.all(), true);
assertEquals(0, results.size());
// query for all in test resource group
- results = paths.getCompactor(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(), true);
+ results = paths.getCompactor(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), true);
assertEquals(1, results.size());
iter = results.iterator();
slp1 = iter.next();
@@ -397,8 +401,8 @@ public class ServiceLockPathsTest {
slp1.toString());
// query for a specific server
- results =
- paths.getCompactor(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.exact(hp), true);
+ results = paths.getCompactor(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.exact(hp), true);
assertEquals(1, results.size());
iter = results.iterator();
slp1 = iter.next();
@@ -410,10 +414,10 @@ public class ServiceLockPathsTest {
// query for a wrong server
for (boolean withLock : new boolean[] {true, false}) {
- results = paths.getCompactor(rg -> rg.equals(TEST_RESOURCE_GROUP),
+ results = paths.getCompactor(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
AddressSelector.exact(HostAndPort.fromString("localhost:1234")), withLock);
assertEquals(0, results.size());
- results = paths.getCompactor(rg -> rg.equals(TEST_RESOURCE_GROUP),
+ results = paths.getCompactor(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
AddressSelector.matching(hp -> hp.equals("localhost:1234")), withLock);
assertEquals(0, results.size());
}
@@ -426,14 +430,13 @@ public class ServiceLockPathsTest {
assertThrows(NullPointerException.class, () -> paths.getScanServer(null, null, true));
assertThrows(NullPointerException.class,
- () -> paths.getScanServer(rg -> rg.equals(TEST_RESOURCE_GROUP), null, true));
- assertTrue(paths.getScanServer(rg -> true, AddressSelector.all(), true).isEmpty());
- assertTrue(
- paths.getScanServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(), true)
- .isEmpty());
+ () -> paths.getScanServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP), null, true));
assertTrue(
- paths.getScanServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.exact(hp), true)
- .isEmpty());
+ paths.getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
+ assertTrue(paths.getScanServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), true).isEmpty());
+ assertTrue(paths.getScanServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.exact(hp), true).isEmpty());
}
@Test
@@ -475,7 +478,8 @@ public class ServiceLockPathsTest {
replay(zc);
// query for all
- Set<ServiceLockPath> results = paths.getScanServer(rg -> true, AddressSelector.all(), false);
+ Set<ServiceLockPath> results =
+ paths.getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
assertEquals(4, results.size());
for (ServiceLockPath path : results) {
assertEquals(ZSSERVERS, path.getType());
@@ -493,7 +497,7 @@ public class ServiceLockPathsTest {
}
// query for all with lock
- results = paths.getScanServer(rg -> true, AddressSelector.all(), true);
+ results = paths.getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
assertEquals(2, results.size());
Iterator<ServiceLockPath> iter = results.iterator();
var slp1 = iter.next();
@@ -522,13 +526,14 @@ public class ServiceLockPathsTest {
}
// query for all in non-existent resource group
- results = paths.getScanServer(rg -> rg.equals(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
- AddressSelector.all(), true);
+ results =
+ paths.getScanServer(ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
+ AddressSelector.all(), true);
assertEquals(0, results.size());
// query for all in test resource group
- results =
- paths.getScanServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(), true);
+ results = paths.getScanServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), true);
assertEquals(1, results.size());
iter = results.iterator();
slp1 = iter.next();
@@ -539,8 +544,8 @@ public class ServiceLockPathsTest {
slp1.toString());
// query for a specific server
- results =
- paths.getScanServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.exact(hp), true);
+ results = paths.getScanServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.exact(hp), true);
assertEquals(1, results.size());
iter = results.iterator();
slp1 = iter.next();
@@ -551,7 +556,7 @@ public class ServiceLockPathsTest {
slp1.toString());
// query for a wrong server
- results = paths.getScanServer(rg -> rg.equals(TEST_RESOURCE_GROUP),
+ results = paths.getScanServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
AddressSelector.exact(HostAndPort.fromString("localhost:1234")), true);
assertEquals(0, results.size());
}
@@ -563,14 +568,13 @@ public class ServiceLockPathsTest {
assertThrows(NullPointerException.class, () -> paths.getTabletServer(null, null, true));
assertThrows(NullPointerException.class,
- () -> paths.getTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), null, true));
- assertTrue(paths.getTabletServer(rg -> true, AddressSelector.all(), true).isEmpty());
+ () -> paths.getTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP), null, true));
assertTrue(
- paths.getTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(), true)
- .isEmpty());
- assertTrue(
- paths.getTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.exact(hp), true)
- .isEmpty());
+ paths.getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
+ assertTrue(paths.getTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), true).isEmpty());
+ assertTrue(paths.getTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.exact(hp), true).isEmpty());
}
@Test
@@ -612,7 +616,8 @@ public class ServiceLockPathsTest {
replay(zc);
// query for all
- Set<ServiceLockPath> results = paths.getTabletServer(rg -> true, AddressSelector.all(), false);
+ Set<ServiceLockPath> results =
+ paths.getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
assertEquals(4, results.size());
for (ServiceLockPath path : results) {
assertEquals(ZTSERVERS, path.getType());
@@ -630,7 +635,7 @@ public class ServiceLockPathsTest {
}
// query for all with lock
- results = paths.getTabletServer(rg -> true, AddressSelector.all(), true);
+ results = paths.getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
assertEquals(2, results.size());
Iterator<ServiceLockPath> iter = results.iterator();
var slp1 = iter.next();
@@ -659,13 +664,14 @@ public class ServiceLockPathsTest {
}
// query for all in non-existent resource group
- results = paths.getTabletServer(rg -> rg.equals(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
+ results = paths.getTabletServer(
+ ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
AddressSelector.all(), true);
assertEquals(0, results.size());
// query for all in test resource group
- results =
- paths.getTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(), true);
+ results = paths.getTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), true);
assertEquals(1, results.size());
iter = results.iterator();
slp1 = iter.next();
@@ -676,8 +682,8 @@ public class ServiceLockPathsTest {
slp1.toString());
// query for a specific server
- results = paths.getTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.exact(hp),
- true);
+ results = paths.getTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.exact(hp), true);
assertEquals(1, results.size());
iter = results.iterator();
slp1 = iter.next();
@@ -688,7 +694,7 @@ public class ServiceLockPathsTest {
slp1.toString());
// query for a wrong server
- results = paths.getTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP),
+ results = paths.getTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
AddressSelector.exact(HostAndPort.fromString("localhost:1234")), true);
assertEquals(0, results.size());
}
@@ -699,15 +705,14 @@ public class ServiceLockPathsTest {
replay(zc);
assertThrows(NullPointerException.class, () -> paths.getDeadTabletServer(null, null, false));
- assertThrows(NullPointerException.class,
- () -> paths.getDeadTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), null, false));
- assertTrue(paths.getDeadTabletServer(rg -> true, AddressSelector.all(), false).isEmpty());
- assertTrue(paths
- .getDeadTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(), false)
- .isEmpty());
- assertTrue(paths
- .getDeadTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.exact(hp), false)
+ assertThrows(NullPointerException.class, () -> paths
+ .getDeadTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP), null, false));
+ assertTrue(paths.getDeadTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false)
.isEmpty());
+ assertTrue(paths.getDeadTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), false).isEmpty());
+ assertTrue(paths.getDeadTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.exact(hp), false).isEmpty());
}
@Test
@@ -746,7 +751,7 @@ public class ServiceLockPathsTest {
// query for all
Set<ServiceLockPath> results =
- paths.getDeadTabletServer(rg -> true, AddressSelector.all(), false);
+ paths.getDeadTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
assertEquals(2, results.size());
Iterator<ServiceLockPath> iter = results.iterator();
var slp1 = iter.next();
@@ -775,13 +780,14 @@ public class ServiceLockPathsTest {
}
// query for all in non-existent resource group
- results = paths.getDeadTabletServer(rg -> rg.equals(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
+ results = paths.getDeadTabletServer(
+ ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE_RESOURCE_GROUP")),
AddressSelector.all(), false);
assertEquals(0, results.size());
// query for all in test resource group
- results = paths.getDeadTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP), AddressSelector.all(),
- false);
+ results = paths.getDeadTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
+ AddressSelector.all(), false);
assertEquals(1, results.size());
iter = results.iterator();
slp1 = iter.next();
@@ -792,7 +798,7 @@ public class ServiceLockPathsTest {
slp1.toString());
// query for a specific server
- results = paths.getDeadTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP),
+ results = paths.getDeadTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
AddressSelector.exact(hp), false);
assertEquals(1, results.size());
iter = results.iterator();
@@ -804,7 +810,7 @@ public class ServiceLockPathsTest {
slp1.toString());
// query for a wrong server
- results = paths.getDeadTabletServer(rg -> rg.equals(TEST_RESOURCE_GROUP),
+ results = paths.getDeadTabletServer(ResourceGroupPredicate.exact(TEST_RESOURCE_GROUP),
AddressSelector.exact(HostAndPort.fromString("localhost:1234")), false);
assertEquals(0, results.size());
}
diff --git a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
index e44e28a069..63aa27227d 100644
--- a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
+++ b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
@@ -86,6 +86,7 @@ import org.apache.accumulo.core.lock.ServiceLock.LockLossReason;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.manager.thrift.ManagerGoalState;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
@@ -850,8 +851,8 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
int tsActualCount = 0;
while (tsActualCount < tsExpectedCount) {
- Set<ServiceLockPath> tservers =
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.all(), true);
+ Set<ServiceLockPath> tservers = context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
tsActualCount = tservers.size();
log.info(tsActualCount + " of " + tsExpectedCount + " tablet servers present in ZooKeeper");
Thread.sleep(500);
@@ -859,8 +860,8 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
int ssActualCount = 0;
while (ssActualCount < ssExpectedCount) {
- Set<ServiceLockPath> tservers =
- context.getServerPaths().getScanServer(rg -> true, AddressSelector.all(), true);
+ Set<ServiceLockPath> tservers = context.getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
ssActualCount = tservers.size();
log.info(ssActualCount + " of " + ssExpectedCount + " scan servers present in ZooKeeper");
Thread.sleep(500);
@@ -868,8 +869,8 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
int ecActualCount = 0;
while (ecActualCount < ecExpectedCount) {
- Set<ServiceLockPath> compactors =
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.all(), true);
+ Set<ServiceLockPath> compactors = context.getServerPaths()
+ .getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
ecActualCount = compactors.size();
log.info(ecActualCount + " of " + ecExpectedCount + " compactors present in ZooKeeper");
Thread.sleep(500);
@@ -1154,7 +1155,8 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
try (AccumuloClient c = Accumulo.newClient().from(clientProperties.get()).build()) {
ClientContext context = (ClientContext) c;
return ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java b/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java
index d367f356eb..194809d454 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/manager/LiveTServerSet.java
@@ -237,8 +237,8 @@ public class LiveTServerSet implements ZooCacheWatcher {
try {
final Set<TServerInstance> updates = new HashSet<>();
final Set<TServerInstance> doomed = new HashSet<>();
- final Set<ServiceLockPath> tservers =
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.all(), false);
+ final Set<ServiceLockPath> tservers = context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
locklessServers.keySet().retainAll(tservers);
@@ -506,7 +506,7 @@ public class LiveTServerSet implements ZooCacheWatcher {
ResourceGroupPredicate rgPredicate = resourceGroup.map(rg -> {
ResourceGroupPredicate rgp = rg2 -> rg.equals(rg2);
return rgp;
- }).orElse(rg -> true);
+ }).orElse(ResourceGroupPredicate.ANY);
AddressSelector addrPredicate =
address.map(AddressSelector::exact).orElse(AddressSelector.all());
Set<ServiceLockPath> paths =
diff --git a/server/base/src/main/java/org/apache/accumulo/server/manager/state/DeadServerList.java b/server/base/src/main/java/org/apache/accumulo/server/manager/state/DeadServerList.java
index 46c3b246d3..03e2a74c88 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/manager/state/DeadServerList.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/manager/state/DeadServerList.java
@@ -30,6 +30,7 @@ import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.manager.thrift.DeadServer;
import org.apache.accumulo.server.ServerContext;
@@ -65,8 +66,8 @@ public class DeadServerList {
public List<DeadServer> getList() {
List<DeadServer> result = new ArrayList<>();
try {
- Set<ServiceLockPath> deadServers =
- ctx.getServerPaths().getDeadTabletServer(rg -> true, AddressSelector.all(), false);
+ Set<ServiceLockPath> deadServers = ctx.getServerPaths()
+ .getDeadTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
for (ServiceLockPath path : deadServers) {
Stat stat = new Stat();
byte[] data;
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/AccumuloStatus.java b/server/base/src/main/java/org/apache/accumulo/server/util/AccumuloStatus.java
index f37fea7d0d..85961b2892 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/AccumuloStatus.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/AccumuloStatus.java
@@ -22,6 +22,7 @@ import java.util.Set;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
public class AccumuloStatus {
@@ -32,8 +33,8 @@ public class AccumuloStatus {
* false
*/
public static boolean isAccumuloOffline(ClientContext context) {
- Set<ServiceLockPath> tservers =
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.all(), true);
+ Set<ServiceLockPath> tservers = context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
if (!tservers.isEmpty()) {
return false;
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java b/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java
index 47518a9b72..c6ad5e2d53 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/Admin.java
@@ -75,6 +75,7 @@ import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.manager.thrift.FateService;
import org.apache.accumulo.core.manager.thrift.TFateId;
@@ -654,7 +655,8 @@ public class Admin implements KeywordExecutable {
throws AccumuloException, AccumuloSecurityException {
ThriftClientTypes.MANAGER.executeVoid(context,
- client -> client.shutdown(TraceUtil.traceInfo(), context.rpcCreds(), tabletServersToo));
+ client -> client.shutdown(TraceUtil.traceInfo(), context.rpcCreds(), tabletServersToo),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
private static void stopServers(final ServerContext context, List<String> servers,
@@ -684,10 +686,12 @@ public class Admin implements KeywordExecutable {
AddressSelector addresses = AddressSelector.matching(hostAndPort::contains);
List<ServiceLockPath> pathsToRemove = new ArrayList<>();
- pathsToRemove.addAll(context.getServerPaths().getCompactor(rg -> true, addresses, false));
- pathsToRemove.addAll(context.getServerPaths().getScanServer(rg -> true, addresses, false));
- pathsToRemove
- .addAll(context.getServerPaths().getTabletServer(rg -> true, addresses, false));
+ pathsToRemove.addAll(
+ context.getServerPaths().getCompactor(ResourceGroupPredicate.ANY, addresses, false));
+ pathsToRemove.addAll(
+ context.getServerPaths().getScanServer(ResourceGroupPredicate.ANY, addresses, false));
+ pathsToRemove.addAll(
+ context.getServerPaths().getTabletServer(ResourceGroupPredicate.ANY, addresses, false));
ZooZap.filterSingleton(context, context.getServerPaths().getManager(false), addresses)
.ifPresent(pathsToRemove::add);
ZooZap.filterSingleton(context, context.getServerPaths().getGarbageCollector(false),
@@ -756,8 +760,11 @@ public class Admin implements KeywordExecutable {
HostAndPort address = AddressUtil.parseAddress(server, port);
final String finalServer = qualifyWithZooKeeperSessionId(context, zc, address.toString());
log.info("Stopping server {}", finalServer);
- ThriftClientTypes.MANAGER.executeVoid(context, client -> client
- .shutdownTabletServer(TraceUtil.traceInfo(), context.rpcCreds(), finalServer, force));
+ ThriftClientTypes.MANAGER
+ .executeVoid(
+ context, client -> client.shutdownTabletServer(TraceUtil.traceInfo(),
+ context.rpcCreds(), finalServer, force),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
}
}
@@ -772,8 +779,8 @@ public class Admin implements KeywordExecutable {
static String qualifyWithZooKeeperSessionId(ClientContext context, ZooCache zooCache,
String hostAndPort) {
var hpObj = HostAndPort.fromString(hostAndPort);
- Set<ServiceLockPath> paths =
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.exact(hpObj), true);
+ Set<ServiceLockPath> paths = context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.exact(hpObj), true);
if (paths.size() != 1) {
return hostAndPort;
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/ServiceStatusCmd.java b/server/base/src/main/java/org/apache/accumulo/server/util/ServiceStatusCmd.java
index ee1bc90e1d..c43492d339 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/ServiceStatusCmd.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/ServiceStatusCmd.java
@@ -31,6 +31,7 @@ import org.apache.accumulo.core.fate.zookeeper.ZooReader;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.server.ServerContext;
@@ -108,8 +109,8 @@ public class ServiceStatusCmd {
final AtomicInteger errors = new AtomicInteger(0);
final Map<String,Set<String>> hostsByGroups = new TreeMap<>();
final Map<String,Integer> resourceGroups = new HashMap<>();
- final Set<ServiceLockPath> compactors =
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.all(), true);
+ final Set<ServiceLockPath> compactors = context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
compactors.forEach(
c -> hostsByGroups.computeIfAbsent(c.getResourceGroup().canonical(), (k) -> new TreeSet<>())
.add(c.getServer()));
@@ -128,8 +129,8 @@ public class ServiceStatusCmd {
final AtomicInteger errors = new AtomicInteger(0);
final Map<String,Set<String>> hostsByGroups = new TreeMap<>();
final Map<String,Integer> resourceGroups = new HashMap<>();
- final Set<ServiceLockPath> scanServers =
- context.getServerPaths().getScanServer(rg -> true, AddressSelector.all(), true);
+ final Set<ServiceLockPath> scanServers = context.getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
scanServers.forEach(
c -> hostsByGroups.computeIfAbsent(c.getResourceGroup().canonical(), (k) -> new TreeSet<>())
.add(c.getServer()));
@@ -158,8 +159,8 @@ public class ServiceStatusCmd {
final AtomicInteger errors = new AtomicInteger(0);
final Map<String,Set<String>> hostsByGroups = new TreeMap<>();
final Map<String,Integer> resourceGroups = new HashMap<>();
- final Set<ServiceLockPath> compactors =
- context.getServerPaths().getCompactor(rg -> true, AddressSelector.all(), true);
+ final Set<ServiceLockPath> compactors = context.getServerPaths()
+ .getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true);
compactors.forEach(
c -> hostsByGroups.computeIfAbsent(c.getResourceGroup().canonical(), (k) -> new TreeSet<>())
.add(c.getServer()));
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java b/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
index 9b427be39a..545be6f88a 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
@@ -26,6 +26,7 @@ import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.zookeeper.ZooCache;
import org.apache.accumulo.server.ServerContext;
@@ -42,8 +43,8 @@ public class TabletServerLocks {
ZooReaderWriter zoo = context.getZooSession().asReaderWriter();
if (delete == null) {
- Set<ServiceLockPath> tabletServers =
- context.getServerPaths().getTabletServer(rg -> true, AddressSelector.all(), false);
+ Set<ServiceLockPath> tabletServers = context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
if (tabletServers.isEmpty()) {
System.err.println("No tservers found in ZK");
}
@@ -64,8 +65,8 @@ public class TabletServerLocks {
printUsage();
} else {
var hostAndPort = HostAndPort.fromString(lock);
- Set<ServiceLockPath> paths = context.getServerPaths().getTabletServer(rg -> true,
- AddressSelector.exact(hostAndPort), true);
+ Set<ServiceLockPath> paths = context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.exact(hostAndPort), true);
Preconditions.checkArgument(paths.size() == 1,
lock + " does not match a single ZooKeeper TabletServer lock. matches=" + paths);
ServiceLockPath path = paths.iterator().next();
diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java b/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java
index ac57cc200b..8b84b2099e 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java
@@ -137,7 +137,7 @@ public class ZooZap implements KeywordExecutable {
.map(ResourceGroupId::of).collect(Collectors.toSet());
rgp = groups::contains;
} else {
- rgp = g -> true;
+ rgp = ResourceGroupPredicate.ANY;
}
if (!opts.zapManager && !opts.zapTservers && !opts.zapCompactors && !opts.zapScanServers
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
index 63fae28c47..a354ea88f8 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/Manager.java
@@ -85,6 +85,7 @@ import org.apache.accumulo.core.lock.ServiceLockData.ServiceDescriptor;
import org.apache.accumulo.core.lock.ServiceLockData.ServiceDescriptors;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.lock.ServiceLockSupport.HAServiceLockWatcher;
import org.apache.accumulo.core.logging.ConditionalLogger.DeduplicatingLogger;
@@ -587,8 +588,8 @@ public class Manager extends AbstractServer implements LiveTServerSet.Listener {
while (stillManager()) {
try {
- Set<ServiceLockPath> scanServerPaths =
- getContext().getServerPaths().getScanServer(rg -> true, AddressSelector.all(), false);
+ Set<ServiceLockPath> scanServerPaths = getContext().getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false);
for (ServiceLockPath path : scanServerPaths) {
ZcStat stat = new ZcStat();
diff --git a/shell/src/main/java/org/apache/accumulo/shell/commands/ListBulkCommand.java b/shell/src/main/java/org/apache/accumulo/shell/commands/ListBulkCommand.java
index 117747c495..44370ca41c 100644
--- a/shell/src/main/java/org/apache/accumulo/shell/commands/ListBulkCommand.java
+++ b/shell/src/main/java/org/apache/accumulo/shell/commands/ListBulkCommand.java
@@ -19,6 +19,7 @@
package org.apache.accumulo.shell.commands;
import org.apache.accumulo.core.clientImpl.ClientContext;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.trace.TraceUtil;
@@ -44,7 +45,8 @@ public class ListBulkCommand extends Command {
ClientContext context = shellState.getContext();
ManagerMonitorInfo stats = ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
final boolean paginate = !cl.hasOption(disablePaginationOpt.getOpt());
diff --git a/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java b/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java
index c7467bb949..d159f1624e 100644
--- a/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java
+++ b/shell/src/main/java/org/apache/accumulo/shell/commands/ListScansCommand.java
@@ -32,6 +32,7 @@ import org.apache.accumulo.core.client.admin.InstanceOperations;
import org.apache.accumulo.core.client.admin.ScanType;
import org.apache.accumulo.core.client.admin.servers.ServerId;
import org.apache.accumulo.core.data.ResourceGroupId;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.util.DurationFormat;
import org.apache.accumulo.shell.Shell;
import org.apache.accumulo.shell.Shell.Command;
@@ -150,7 +151,7 @@ public class ListScansCommand extends Command {
static Predicate<ResourceGroupId> rgRegexPredicate(String rgRegex) {
if (rgRegex == null || rgRegex.isBlank()) {
- return rgid -> true;
+ return ResourceGroupPredicate.ANY;
}
final Pattern p = Pattern.compile(rgRegex);
return rgid -> p.matcher(rgid.canonical()).matches();
diff --git a/test/src/main/java/org/apache/accumulo/test/DetectDeadTabletServersIT.java b/test/src/main/java/org/apache/accumulo/test/DetectDeadTabletServersIT.java
index 36e0faaec1..afe745e8fa 100644
--- a/test/src/main/java/org/apache/accumulo/test/DetectDeadTabletServersIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/DetectDeadTabletServersIT.java
@@ -28,6 +28,7 @@ import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.ClientProperty;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
@@ -81,7 +82,8 @@ public class DetectDeadTabletServersIT extends ConfigurableMacBase {
private ManagerMonitorInfo getStats(AccumuloClient c) throws Exception {
final ClientContext context = (ClientContext) c;
return ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
}
diff --git a/test/src/main/java/org/apache/accumulo/test/GetManagerStats.java b/test/src/main/java/org/apache/accumulo/test/GetManagerStats.java
index 21ff2e03a9..e069225a2a 100644
--- a/test/src/main/java/org/apache/accumulo/test/GetManagerStats.java
+++ b/test/src/main/java/org/apache/accumulo/test/GetManagerStats.java
@@ -23,6 +23,7 @@ import java.util.Date;
import java.util.Map.Entry;
import org.apache.accumulo.core.conf.SiteConfiguration;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.BulkImportStatus;
import org.apache.accumulo.core.manager.thrift.DeadServer;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
@@ -39,7 +40,8 @@ public class GetManagerStats {
ManagerMonitorInfo stats = null;
var context = new ServerContext(SiteConfiguration.auto());
stats = ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
out(0, "State: " + stats.state.name());
out(0, "Goal State: " + stats.goalState.name());
if (stats.serversShuttingDown != null && !stats.serversShuttingDown.isEmpty()) {
diff --git a/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java b/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
index 1a05a773d0..0c3600eac9 100644
--- a/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
@@ -39,6 +39,7 @@ import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
import org.apache.accumulo.core.spi.balancer.TableLoadBalancer;
@@ -143,12 +144,16 @@ public class RecoveryIT extends AccumuloClusterHarness {
// Stop any running Compactors and ScanServers
control.stopAllServers(ServerType.COMPACTOR);
- Wait.waitFor(() -> getServerContext().getServerPaths()
- .getCompactor(rg -> true, AddressSelector.all(), true).isEmpty(), 60_000);
+ Wait.waitFor(
+ () -> getServerContext().getServerPaths()
+ .getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty(),
+ 60_000);
control.stopAllServers(ServerType.SCAN_SERVER);
- Wait.waitFor(() -> getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).size() == 0, 60_000);
+ Wait.waitFor(
+ () -> getServerContext().getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 0,
+ 60_000);
// Kill the TabletServer in resource group that is hosting the table
List<Process> procs = control.getTabletServers(RESOURCE_GROUP);
diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerConcurrentTabletScanIT.java b/test/src/main/java/org/apache/accumulo/test/ScanServerConcurrentTabletScanIT.java
index 170355106c..52db3b7831 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerConcurrentTabletScanIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerConcurrentTabletScanIT.java
@@ -40,6 +40,7 @@ import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
@@ -91,7 +92,7 @@ public class ScanServerConcurrentTabletScanIT extends SharedMiniClusterBase {
1, null);
Wait.waitFor(() -> !getCluster().getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).isEmpty());
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
}
diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerGroupConfigurationIT.java b/test/src/main/java/org/apache/accumulo/test/ScanServerGroupConfigurationIT.java
index 7c8683cd80..b5ea2498e2 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerGroupConfigurationIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerGroupConfigurationIT.java
@@ -31,6 +31,7 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
@@ -123,7 +124,7 @@ public class ScanServerGroupConfigurationIT extends SharedMiniClusterBase {
// Ensure no scan servers running
Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).isEmpty());
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
final String tableName = getUniqueNames(1)[0];
@@ -144,13 +145,13 @@ public class ScanServerGroupConfigurationIT extends SharedMiniClusterBase {
// Start a ScanServer. No group specified, should be in the default group.
getCluster().getClusterControl().start(ServerType.SCAN_SERVER, "localhost");
- Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).size() == 1, 30_000);
Wait.waitFor(
- () -> getCluster()
- .getServerContext().getServerPaths().getScanServer(
- rg -> rg.equals(ResourceGroupId.DEFAULT), AddressSelector.all(), true)
- .size() > 0);
+ () -> getCluster().getServerContext().getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 1,
+ 30_000);
+ Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
+ .getScanServer(ResourceGroupPredicate.DEFAULT_RG_ONLY, AddressSelector.all(), true)
+ .size() > 0);
assertEquals(ingestedEntryCount, scanner.stream().count(),
"The scan server scanner should have seen all ingested and flushed entries");
@@ -164,16 +165,16 @@ public class ScanServerGroupConfigurationIT extends SharedMiniClusterBase {
getCluster().getConfig().getClusterServerConfiguration()
.addScanServerResourceGroup("GROUP1", 1);
getCluster().getClusterControl().start(ServerType.SCAN_SERVER);
- Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).size() == 2, 30_000);
Wait.waitFor(
- () -> getCluster()
- .getServerContext().getServerPaths().getScanServer(
- rg -> rg.equals(ResourceGroupId.DEFAULT), AddressSelector.all(), true)
- .size() == 1);
- Wait.waitFor(() -> getCluster()
- .getServerContext().getServerPaths().getScanServer(
- rg -> rg.equals(ResourceGroupId.of("GROUP1")), AddressSelector.all(), true)
+ () -> getCluster().getServerContext().getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 2,
+ 30_000);
+ Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
+ .getScanServer(ResourceGroupPredicate.DEFAULT_RG_ONLY, AddressSelector.all(), true)
+ .size() == 1);
+ Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
+ .getScanServer(ResourceGroupPredicate.exact(ResourceGroupId.of("GROUP1")),
+ AddressSelector.all(), true)
.size() == 1);
scanner.setExecutionHints(Map.of("scan_type", "use_group1"));
diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java b/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java
index 6c80dc4f82..03c228b223 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerIT.java
@@ -60,6 +60,7 @@ import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.fate.FateId;
import org.apache.accumulo.core.fate.FateInstanceType;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.metadata.schema.Ample;
import org.apache.accumulo.core.metadata.schema.Ample.TabletsMutator;
@@ -114,7 +115,7 @@ public class ScanServerIT extends SharedMiniClusterBase {
"localhost");
Wait.waitFor(() -> !getCluster().getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).isEmpty());
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
}
@AfterAll
diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java b/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java
index 8fc3881bbf..78fb9ab44b 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java
@@ -47,6 +47,7 @@ import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.gc.Reference;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.ScanServerRefTabletFile;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
@@ -92,7 +93,7 @@ public class ScanServerMetadataEntriesIT extends SharedMiniClusterBase {
"localhost");
Wait.waitFor(() -> !getCluster().getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).isEmpty());
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
}
diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerMultipleScansIT.java b/test/src/main/java/org/apache/accumulo/test/ScanServerMultipleScansIT.java
index 776957d6c9..735dd7f7d0 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerMultipleScansIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerMultipleScansIT.java
@@ -47,6 +47,7 @@ import org.apache.accumulo.core.client.admin.NewTableConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
@@ -88,7 +89,7 @@ public class ScanServerMultipleScansIT extends SharedMiniClusterBase {
"localhost");
Wait.waitFor(() -> !getCluster().getServerContext().getServerPaths()
- .getScanServer(rg -> true, AddressSelector.all(), true).isEmpty());
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
}
@AfterAll
diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerShutdownIT.java b/test/src/main/java/org/apache/accumulo/test/ScanServerShutdownIT.java
index ed6238f759..5cfa014d8b 100644
--- a/test/src/main/java/org/apache/accumulo/test/ScanServerShutdownIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ScanServerShutdownIT.java
@@ -36,6 +36,7 @@ import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
@@ -86,8 +87,8 @@ public class ScanServerShutdownIT extends SharedMiniClusterBase {
ServerContext ctx = getCluster().getServerContext();
- Wait.waitFor(() -> !ctx.getServerPaths().getScanServer(rg -> true, AddressSelector.all(), true)
- .isEmpty());
+ Wait.waitFor(() -> !ctx.getServerPaths()
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty());
try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
final String tableName = getUniqueNames(1)[0];
diff --git a/test/src/main/java/org/apache/accumulo/test/TransportCachingIT.java b/test/src/main/java/org/apache/accumulo/test/TransportCachingIT.java
index b5e41e4139..d7293c42e5 100644
--- a/test/src/main/java/org/apache/accumulo/test/TransportCachingIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/TransportCachingIT.java
@@ -20,6 +20,7 @@ package org.apache.accumulo.test;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import java.util.List;
@@ -33,6 +34,9 @@ import org.apache.accumulo.core.clientImpl.ThriftTransportKey;
import org.apache.accumulo.core.clientImpl.ThriftTransportPool;
import org.apache.accumulo.core.conf.ConfigurationTypeHelper;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.ResourceGroupId;
+import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.harness.AccumuloClusterHarness;
@@ -72,48 +76,62 @@ public class TransportCachingIT extends AccumuloClusterHarness {
ThriftTransportKey ttk = servers.get(0);
ThriftTransportPool pool = context.getTransportPool();
- TTransport first = getAnyTransport(ttk, pool, true);
+ TTransport first = getAnyTransport(ttk, pool, context, ThriftService.CLIENT,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, true);
assertNotNull(first);
// Return it to unreserve it
pool.returnTransport(first);
- TTransport second = getAnyTransport(ttk, pool, true);
+ TTransport second = getAnyTransport(ttk, pool, context, ThriftService.CLIENT,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, true);
// We should get the same transport
assertSame(first, second, "Expected the first and second to be the same instance");
pool.returnTransport(second);
// Ensure does not get cached connection just returned
- TTransport third = getAnyTransport(ttk, pool, false);
+ TTransport third = getAnyTransport(ttk, pool, context, ThriftService.CLIENT,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, false);
assertNotSame(second, third, "Expected second and third transport to be different instances");
- TTransport fourth = getAnyTransport(ttk, pool, false);
+ TTransport fourth = getAnyTransport(ttk, pool, context, ThriftService.CLIENT,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, false);
assertNotSame(third, fourth, "Expected third and fourth transport to be different instances");
pool.returnTransport(third);
pool.returnTransport(fourth);
// The following three asserts ensure the per server queue is LIFO
- TTransport fifth = getAnyTransport(ttk, pool, true);
+ TTransport fifth = getAnyTransport(ttk, pool, context, ThriftService.CLIENT,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, true);
assertSame(fourth, fifth, "Expected fourth and fifth transport to be the same instance");
- TTransport sixth = getAnyTransport(ttk, pool, true);
+ TTransport sixth = getAnyTransport(ttk, pool, context, ThriftService.CLIENT,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, true);
assertSame(third, sixth, "Expected third and sixth transport to be the same instance");
- TTransport seventh = getAnyTransport(ttk, pool, true);
+ TTransport seventh = getAnyTransport(ttk, pool, context, ThriftService.CLIENT,
+ ResourceGroupPredicate.DEFAULT_RG_ONLY, true);
assertSame(second, seventh, "Expected second and seventh transport to be the same instance");
pool.returnTransport(fifth);
pool.returnTransport(sixth);
pool.returnTransport(seventh);
+
+ Pair<String,TTransport> eigth = pool.getAnyCachedTransport(ttk.getType(), context,
+ ThriftService.CLIENT, ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE")));
+ assertNull(eigth);
+
}
}
private TTransport getAnyTransport(ThriftTransportKey ttk, ThriftTransportPool pool,
+ ClientContext context, ThriftService service, ResourceGroupPredicate rgp,
boolean preferCached) {
if (preferCached) {
- Pair<String,TTransport> cached = pool.getAnyCachedTransport(ttk.getType());
+ Pair<String,TTransport> cached =
+ pool.getAnyCachedTransport(ttk.getType(), context, service, rgp);
if (cached != null) {
return cached.getSecond();
}
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
index 7ba8812ef4..3edcc01715 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
@@ -56,6 +56,7 @@ import org.apache.accumulo.core.file.FileOperations;
import org.apache.accumulo.core.file.FileSKVWriter;
import org.apache.accumulo.core.file.rfile.RFile;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.UnreferencedTabletFile;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
@@ -118,8 +119,10 @@ public class CompactionPriorityQueueMetricsIT extends SharedMiniClusterBase {
@BeforeEach
public void setupMetricsTest() throws Exception {
getCluster().getClusterControl().stopAllServers(ServerType.COMPACTOR);
- Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
- .getCompactor(rg -> true, AddressSelector.all(), true).isEmpty(), 60_000);
+ Wait.waitFor(
+ () -> getCluster().getServerContext().getServerPaths()
+ .getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty(),
+ 60_000);
try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
tableName = getUniqueNames(1)[0];
diff --git a/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java b/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java
index da2f0272b5..006aa96d15 100644
--- a/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java
+++ b/test/src/main/java/org/apache/accumulo/test/conf/PropStoreConfigIT_SimpleSuite.java
@@ -46,6 +46,7 @@ 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.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.harness.SharedMiniClusterBase;
@@ -483,10 +484,9 @@ public class PropStoreConfigIT_SimpleSuite extends SharedMiniClusterBase {
private Map<String,String> getStoredConfiguration() throws Exception {
ServerContext ctx = getCluster().getServerContext();
- return ThriftClientTypes.CLIENT
- .execute(ctx,
- client -> client.getVersionedSystemProperties(TraceUtil.traceInfo(), ctx.rpcCreds()))
- .getProperties();
+ return ThriftClientTypes.CLIENT.execute(ctx,
+ client -> client.getVersionedSystemProperties(TraceUtil.traceInfo(), ctx.rpcCreds()),
+ ResourceGroupPredicate.ANY).getProperties();
}
interface PropertyShim {
diff --git a/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java b/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java
index 853b9ab9e8..0f62184103 100644
--- a/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java
+++ b/test/src/main/java/org/apache/accumulo/test/fate/FateOpsCommandsITBase.java
@@ -66,6 +66,7 @@ import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.iterators.IteratorUtil;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.harness.SharedMiniClusterBase;
@@ -104,8 +105,10 @@ public abstract class FateOpsCommandsITBase extends SharedMiniClusterBase
// initiated on starting the manager, causing the test to fail. Stopping the compactor fixes
// this issue.
getCluster().getClusterControl().stopAllServers(ServerType.COMPACTOR);
- Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
- .getCompactor(rg -> true, AddressSelector.all(), true).isEmpty(), 60_000);
+ Wait.waitFor(
+ () -> getCluster().getServerContext().getServerPaths()
+ .getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).isEmpty(),
+ 60_000);
}
@AfterAll
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BalanceAfterCommsFailureIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BalanceAfterCommsFailureIT.java
index f6688cd567..fb8fdeb41d 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BalanceAfterCommsFailureIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BalanceAfterCommsFailureIT.java
@@ -33,6 +33,7 @@ import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
import org.apache.accumulo.core.manager.thrift.TableInfo;
import org.apache.accumulo.core.manager.thrift.TabletServerStatus;
@@ -105,7 +106,8 @@ public class BalanceAfterCommsFailureIT extends ConfigurableMacBase {
int unassignedTablets = 1;
for (int i = 0; unassignedTablets > 0 && i < 10; i++) {
stats = ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
unassignedTablets = stats.getUnassignedTablets();
if (unassignedTablets > 0) {
log.info("Found {} unassigned tablets, sleeping 3 seconds for tablet assignment",
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BalanceInPresenceOfOfflineTableIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BalanceInPresenceOfOfflineTableIT.java
index 1a0276a980..eddbd52d9f 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BalanceInPresenceOfOfflineTableIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BalanceInPresenceOfOfflineTableIT.java
@@ -40,6 +40,7 @@ import org.apache.accumulo.core.client.admin.servers.ServerId;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.clientImpl.Credentials;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
import org.apache.accumulo.core.manager.thrift.TableInfo;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
@@ -158,7 +159,8 @@ public class BalanceInPresenceOfOfflineTableIT extends AccumuloClusterHarness {
ManagerMonitorInfo stats = ThriftClientTypes.MANAGER.execute((ClientContext) accumuloClient,
client -> client.getManagerStats(TraceUtil.traceInfo(),
- creds.toThrift(accumuloClient.instanceOperations().getInstanceId())));
+ creds.toThrift(accumuloClient.instanceOperations().getInstanceId())),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
if (stats.getTServerInfoSize() < 2) {
log.debug("we need >= 2 servers. sleeping for {}ms", currentWait);
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/DebugClientConnectionIT.java b/test/src/main/java/org/apache/accumulo/test/functional/DebugClientConnectionIT.java
index 84699dde5e..275d7e8e09 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/DebugClientConnectionIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/DebugClientConnectionIT.java
@@ -21,6 +21,7 @@ package org.apache.accumulo.test.functional;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.UncheckedIOException;
import java.util.Iterator;
@@ -28,10 +29,17 @@ import java.util.Set;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.admin.servers.ServerId;
+import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
+import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.rpc.clients.TServerClient;
import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.util.Wait;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -52,6 +60,8 @@ public class DebugClientConnectionIT extends AccumuloClusterHarness {
}
assertNotNull(tservers);
assertEquals(2, tservers.size());
+ System.clearProperty(TServerClient.DEBUG_HOST);
+ System.clearProperty(TServerClient.DEBUG_RG);
}
@Test
@@ -71,4 +81,50 @@ public class DebugClientConnectionIT extends AccumuloClusterHarness {
() -> client.instanceOperations().getSiteConfiguration());
}
}
+
+ @Test
+ public void testPreferredResourceGroup() throws Exception {
+ System.setProperty(TServerClient.DEBUG_RG, "fake");
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ AccumuloException ae = assertThrows(AccumuloException.class,
+ () -> client.tableOperations().getConfiguration(SystemTables.METADATA.tableName()));
+ // TableOps wraps RuntimeExceptions with AccumuloExceptions
+ assertTrue(ae.getCause() instanceof UncheckedIOException);
+ }
+ // Set DEBUG HOST with DEBUG RG, should be no overlap with running hosts
+ Iterator<ServerId> tsi = tservers.iterator();
+ System.setProperty(TServerClient.DEBUG_HOST, tsi.next().toHostPortString());
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ AccumuloException ae = assertThrows(AccumuloException.class,
+ () -> client.tableOperations().getConfiguration(SystemTables.METADATA.tableName()));
+ // TableOps wraps RuntimeExceptions with AccumuloExceptions
+ assertTrue(ae.getCause() instanceof UncheckedIOException);
+ }
+ MiniAccumuloClusterImpl mini = (MiniAccumuloClusterImpl) getCluster();
+ mini.getConfig().getClusterServerConfiguration().addTabletServerResourceGroup("fake", 1);
+ mini.getClusterControl().start(ServerType.TABLET_SERVER);
+ Wait.waitFor(() -> getCluster().getServerContext().getServerPaths()
+ .getTabletServer(r -> r.canonical().equals("fake"), AddressSelector.all(), true).size()
+ == 1);
+
+ System.clearProperty(TServerClient.DEBUG_HOST);
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ assertNotNull(client.tableOperations().getConfiguration(SystemTables.METADATA.tableName()));
+ }
+ // Set DEBUG HOST with DEBUG RG, should be no overlap with running hosts
+ System.setProperty(TServerClient.DEBUG_HOST, tsi.next().toHostPortString());
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ AccumuloException ae = assertThrows(AccumuloException.class,
+ () -> client.tableOperations().getConfiguration(SystemTables.METADATA.tableName()));
+ // TableOps wraps RuntimeExceptions with AccumuloExceptions
+ assertTrue(ae.getCause() instanceof UncheckedIOException);
+ }
+ // Set DEBUG_HOST and DEBUG_RG to overlap
+ Set<ServiceLockPath> fakeTservers = getCluster().getServerContext().getServerPaths()
+ .getTabletServer(r -> r.canonical().equals("fake"), AddressSelector.all(), true);
+ System.setProperty(TServerClient.DEBUG_HOST, fakeTservers.iterator().next().getServer());
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ assertNotNull(client.tableOperations().getConfiguration(SystemTables.METADATA.tableName()));
+ }
+ }
}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/GracefulShutdownIT.java b/test/src/main/java/org/apache/accumulo/test/functional/GracefulShutdownIT.java
index ada20ee904..c1bad36ae4 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/GracefulShutdownIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/GracefulShutdownIT.java
@@ -50,6 +50,7 @@ import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockData;
import org.apache.accumulo.core.lock.ServiceLockData.ThriftService;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
import org.apache.accumulo.core.security.Authorizations;
@@ -220,9 +221,9 @@ public class GracefulShutdownIT extends SharedMiniClusterBase {
.getServerContext().getServerPaths().getCompactor(
(rg) -> rg.equals(ResourceGroupId.of(GROUP_NAME)), AddressSelector.all(), true)
.size() == 1);
- final Set<ServiceLockPath> compactors =
- getCluster().getServerContext().getServerPaths().getCompactor(
- (rg) -> rg.equals(ResourceGroupId.of(GROUP_NAME)), AddressSelector.all(), true);
+ final Set<ServiceLockPath> compactors = getCluster().getServerContext().getServerPaths()
+ .getCompactor(ResourceGroupPredicate.exact(ResourceGroupId.of(GROUP_NAME)),
+ AddressSelector.all(), true);
final HostAndPort compactorAddress =
HostAndPort.fromString(compactors.iterator().next().getServer());
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ManagerApiIT.java b/test/src/main/java/org/apache/accumulo/test/functional/ManagerApiIT.java
index 593f382dfd..9ea4778c49 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ManagerApiIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ManagerApiIT.java
@@ -38,6 +38,7 @@ import org.apache.accumulo.core.clientImpl.Credentials;
import org.apache.accumulo.core.clientImpl.thrift.TVersionedProperties;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.InstanceId;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerClientService;
import org.apache.accumulo.core.manager.thrift.ManagerGoalState;
import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
@@ -266,7 +267,8 @@ public class ManagerApiIT extends SharedMiniClusterBase {
try (AccumuloClient client = Accumulo.newClient().from(getClientProps())
.as(rootUser.getPrincipal(), rootUser.getToken()).build()) {
ClientContext context = (ClientContext) client;
- ThriftClientTypes.MANAGER.execute(context, op.apply(rootUser));
+ ThriftClientTypes.MANAGER.execute(context, op.apply(rootUser),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
}
@@ -303,14 +305,15 @@ public class ManagerApiIT extends SharedMiniClusterBase {
Credentials user) throws Exception {
try (AccumuloClient client = Accumulo.newClient().from(getClientProps())
.as(user.getPrincipal(), user.getToken()).build()) {
- ThriftClientTypes.MANAGER.execute((ClientContext) client, op.apply(user));
+ ThriftClientTypes.MANAGER.execute((ClientContext) client, op.apply(user),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
}
private static void expectPermissionSuccess(
ThriftClientTypes.Exec<Void,ManagerClientService.Client> op, ClientContext context)
throws Exception {
- ThriftClientTypes.MANAGER.execute(context, op);
+ ThriftClientTypes.MANAGER.execute(context, op, ResourceGroupPredicate.DEFAULT_RG_ONLY);
}
private static void expectPermissionDenied(
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java b/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java
index 95bd336da5..dbeb1a9908 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/ManagerAssignmentIT.java
@@ -65,6 +65,7 @@ import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.fate.FateId;
import org.apache.accumulo.core.fate.FateInstanceType;
import org.apache.accumulo.core.lock.ServiceLock;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.metadata.schema.Ample;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
@@ -456,7 +457,8 @@ public class ManagerAssignmentIT extends SharedMiniClusterBase {
public static List<TabletStats> getTabletStats(AccumuloClient c, String tableId)
throws AccumuloException, AccumuloSecurityException {
return ThriftClientTypes.TABLET_SERVER.execute((ClientContext) c, client -> client
- .getTabletStats(TraceUtil.traceInfo(), ((ClientContext) c).rpcCreds(), tableId));
+ .getTabletStats(TraceUtil.traceInfo(), ((ClientContext) c).rpcCreds(), tableId),
+ ResourceGroupPredicate.ANY);
}
@Test
@@ -534,7 +536,8 @@ public class ManagerAssignmentIT extends SharedMiniClusterBase {
try {
ThriftClientTypes.MANAGER.executeVoid((ClientContext) client,
c -> c.shutdownTabletServer(TraceUtil.traceInfo(),
- getCluster().getServerContext().rpcCreds(), finalAddress, false));
+ getCluster().getServerContext().rpcCreds(), finalAddress, false),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
} catch (AccumuloException | AccumuloSecurityException e) {
fail("Error shutting down TabletServer", e);
}
@@ -585,7 +588,8 @@ public class ManagerAssignmentIT extends SharedMiniClusterBase {
try {
ThriftClientTypes.MANAGER.executeVoid((ClientContext) client,
c -> c.shutdownTabletServer(TraceUtil.traceInfo(),
- getCluster().getServerContext().rpcCreds(), finalAddress, false));
+ getCluster().getServerContext().rpcCreds(), finalAddress, false),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
} catch (AccumuloException | AccumuloSecurityException e) {
fail("Error shutting down TabletServer", e);
}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java b/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java
index 3c42b98fc7..8c68a2597b 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/MemoryStarvedMajCIT.java
@@ -37,8 +37,8 @@ import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.admin.servers.ServerId;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
@@ -141,8 +141,8 @@ public class MemoryStarvedMajCIT extends SharedMiniClusterBase {
ClientContext ctx = (ClientContext) client;
Wait.waitFor(() -> ctx.getServerPaths()
- .getCompactor(rg -> rg.equals(ResourceGroupId.DEFAULT), AddressSelector.all(), true)
- .size() == 1, 60_000);
+ .getCompactor(ResourceGroupPredicate.DEFAULT_RG_ONLY, AddressSelector.all(), true).size()
+ == 1, 60_000);
ServerId csi = ctx.instanceOperations().getServers(ServerId.Type.COMPACTOR).iterator().next();
HostAndPort compactorAddr = HostAndPort.fromParts(csi.getHost(), csi.getPort());
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/MetadataMaxFilesIT.java b/test/src/main/java/org/apache/accumulo/test/functional/MetadataMaxFilesIT.java
index 387bc9db7a..92fe15939c 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/MetadataMaxFilesIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/MetadataMaxFilesIT.java
@@ -31,6 +31,7 @@ import org.apache.accumulo.core.client.admin.NewTableConfiguration;
import org.apache.accumulo.core.client.admin.TabletAvailability;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
import org.apache.accumulo.core.manager.thrift.TableInfo;
import org.apache.accumulo.core.manager.thrift.TabletServerStatus;
@@ -82,7 +83,8 @@ public class MetadataMaxFilesIT extends ConfigurableMacBase {
while (true) {
ClientContext context = (ClientContext) c;
ManagerMonitorInfo stats = ThriftClientTypes.MANAGER.execute(context,
- client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()));
+ client -> client.getManagerStats(TraceUtil.traceInfo(), context.rpcCreds()),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
int tablets = 0;
for (TabletServerStatus tserver : stats.tServerInfo) {
for (Entry<String,TableInfo> entry : tserver.tableMap.entrySet()) {
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/SimpleBalancerFairnessIT.java b/test/src/main/java/org/apache/accumulo/test/functional/SimpleBalancerFairnessIT.java
index 65b11013e8..fbbd019465 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/SimpleBalancerFairnessIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/SimpleBalancerFairnessIT.java
@@ -35,6 +35,7 @@ import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.clientImpl.Credentials;
import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.manager.thrift.ManagerMonitorInfo;
import org.apache.accumulo.core.manager.thrift.TableInfo;
import org.apache.accumulo.core.manager.thrift.TabletServerStatus;
@@ -87,7 +88,8 @@ public class SimpleBalancerFairnessIT extends ConfigurableMacBase {
Wait.waitFor(() -> {
ManagerMonitorInfo stats = ThriftClientTypes.MANAGER.execute(context,
client -> client.getManagerStats(TraceUtil.traceInfo(),
- creds.toThrift(c.instanceOperations().getInstanceId())));
+ creds.toThrift(c.instanceOperations().getInstanceId())),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
int unassignedTablets = stats.getUnassignedTablets();
if (unassignedTablets > 0) {
log.info("Found {} unassigned tablets, sleeping 3 seconds for tablet assignment",
@@ -102,7 +104,8 @@ public class SimpleBalancerFairnessIT extends ConfigurableMacBase {
Wait.waitFor(() -> {
ManagerMonitorInfo stats = ThriftClientTypes.MANAGER.execute(context,
client -> client.getManagerStats(TraceUtil.traceInfo(),
- creds.toThrift(c.instanceOperations().getInstanceId())));
+ creds.toThrift(c.instanceOperations().getInstanceId())),
+ ResourceGroupPredicate.DEFAULT_RG_ONLY);
List<Integer> counts = new ArrayList<>();
for (TabletServerStatus server : stats.tServerInfo) {
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java b/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java
index 666c0e5149..1042e7b5fe 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/TabletManagementIteratorIT.java
@@ -72,6 +72,7 @@ import org.apache.accumulo.core.fate.FateId;
import org.apache.accumulo.core.fate.FateInstanceType;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.manager.state.TabletManagement;
import org.apache.accumulo.core.manager.state.tables.TableState;
@@ -690,8 +691,8 @@ public class TabletManagementIteratorIT extends AccumuloClusterHarness {
tableId -> context.getTableState(tableId) == TableState.ONLINE);
HashSet<TServerInstance> tservers = new HashSet<>();
- for (ServiceLockPath tserver : context.getServerPaths().getTabletServer(rg -> true,
- AddressSelector.all(), true)) {
+ for (ServiceLockPath tserver : context.getServerPaths()
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true)) {
try {
long sessionId = ServiceLock.getSessionId(context.getZooCache(), tserver);
tservers.add(new TServerInstance(tserver.getServer(), sessionId));
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/TabletResourceGroupBalanceIT.java b/test/src/main/java/org/apache/accumulo/test/functional/TabletResourceGroupBalanceIT.java
index dc965919c8..0f44b42335 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/TabletResourceGroupBalanceIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/TabletResourceGroupBalanceIT.java
@@ -47,6 +47,7 @@ import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.metadata.schema.Ample;
@@ -101,7 +102,7 @@ public class TabletResourceGroupBalanceIT extends SharedMiniClusterBase {
Map<String,ResourceGroupId> tservers = new HashMap<>();
for (ServiceLockPath tserver : cluster.getServerContext().getServerPaths()
- .getTabletServer(rg -> true, AddressSelector.all(), true)) {
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true)) {
tservers.put(tserver.getServer(), tserver.getResourceGroup());
}
return tservers;
diff --git a/test/src/main/java/org/apache/accumulo/test/lock/ServiceLockPathsIT.java b/test/src/main/java/org/apache/accumulo/test/lock/ServiceLockPathsIT.java
index 7475abf5b5..d35c98b5ca 100644
--- a/test/src/main/java/org/apache/accumulo/test/lock/ServiceLockPathsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/lock/ServiceLockPathsIT.java
@@ -26,6 +26,7 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.lock.ServiceLockPaths;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.harness.AccumuloClusterHarness;
import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
@@ -52,68 +53,57 @@ public class ServiceLockPathsIT extends AccumuloClusterHarness {
assertNotNull(paths.getGarbageCollector(true));
assertNotNull(paths.getManager(true));
assertNull(paths.getMonitor(true)); // monitor not started
- assertEquals(2, paths.getTabletServer(rg -> true, AddressSelector.all(), true).size());
+ assertEquals(2,
+ paths.getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size());
assertEquals(1,
- paths.getTabletServer(rg -> rg.equals(ResourceGroupId.DEFAULT), AddressSelector.all(), true)
- .size());
- assertEquals(1, paths
- .getTabletServer(rg -> rg.equals(ResourceGroupId.of("TTEST")), AddressSelector.all(), true)
- .size());
- assertEquals(0, paths
- .getTabletServer(rg -> rg.equals(ResourceGroupId.of("FAKE")), AddressSelector.all(), true)
- .size());
- assertEquals(0, paths
- .getTabletServer(rg -> rg.equals(ResourceGroupId.of("CTEST")), AddressSelector.all(), true)
- .size());
- assertEquals(0, paths
- .getTabletServer(rg -> rg.equals(ResourceGroupId.of("STEST")), AddressSelector.all(), true)
- .size());
-
- assertEquals(4, paths.getCompactor(rg -> true, AddressSelector.all(), true).size());
- assertEquals(1,
- paths.getCompactor(rg -> rg.equals(ResourceGroupId.DEFAULT), AddressSelector.all(), true)
- .size());
- assertEquals(3,
- paths
- .getCompactor(rg -> rg.equals(ResourceGroupId.of("CTEST")), AddressSelector.all(), true)
- .size());
- assertEquals(0,
- paths.getCompactor(rg -> rg.equals(ResourceGroupId.of("FAKE")), AddressSelector.all(), true)
- .size());
- assertEquals(0,
- paths
- .getCompactor(rg -> rg.equals(ResourceGroupId.of("TTEST")), AddressSelector.all(), true)
- .size());
- assertEquals(0,
- paths
- .getCompactor(rg -> rg.equals(ResourceGroupId.of("STEST")), AddressSelector.all(), true)
+ paths.getTabletServer(ResourceGroupPredicate.DEFAULT_RG_ONLY, AddressSelector.all(), true)
.size());
+ assertEquals(1, paths.getTabletServer(ResourceGroupPredicate.exact(ResourceGroupId.of("TTEST")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getTabletServer(ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getTabletServer(ResourceGroupPredicate.exact(ResourceGroupId.of("CTEST")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getTabletServer(ResourceGroupPredicate.exact(ResourceGroupId.of("STEST")),
+ AddressSelector.all(), true).size());
+
+ assertEquals(4,
+ paths.getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size());
+ assertEquals(1, paths.getCompactor(ResourceGroupPredicate.exact(ResourceGroupId.DEFAULT),
+ AddressSelector.all(), true).size());
+ assertEquals(3, paths.getCompactor(ResourceGroupPredicate.exact(ResourceGroupId.of("CTEST")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getCompactor(ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getCompactor(ResourceGroupPredicate.exact(ResourceGroupId.of("TTEST")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getCompactor(ResourceGroupPredicate.exact(ResourceGroupId.of("STEST")),
+ AddressSelector.all(), true).size());
- assertEquals(3, paths.getScanServer(rg -> true, AddressSelector.all(), true).size());
- assertEquals(1,
- paths.getScanServer(rg -> rg.equals(ResourceGroupId.DEFAULT), AddressSelector.all(), true)
- .size());
- assertEquals(2, paths
- .getScanServer(rg -> rg.equals(ResourceGroupId.of("STEST")), AddressSelector.all(), true)
- .size());
- assertEquals(0,
- paths
- .getScanServer(rg -> rg.equals(ResourceGroupId.of("FAKE")), AddressSelector.all(), true)
- .size());
- assertEquals(0, paths
- .getScanServer(rg -> rg.equals(ResourceGroupId.of("CTEST")), AddressSelector.all(), true)
- .size());
- assertEquals(0, paths
- .getScanServer(rg -> rg.equals(ResourceGroupId.of("TTEST")), AddressSelector.all(), true)
- .size());
+ assertEquals(3,
+ paths.getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size());
+ assertEquals(1, paths
+ .getScanServer(ResourceGroupPredicate.DEFAULT_RG_ONLY, AddressSelector.all(), true).size());
+ assertEquals(2, paths.getScanServer(ResourceGroupPredicate.exact(ResourceGroupId.of("STEST")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getScanServer(ResourceGroupPredicate.exact(ResourceGroupId.of("FAKE")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getScanServer(ResourceGroupPredicate.exact(ResourceGroupId.of("CTEST")),
+ AddressSelector.all(), true).size());
+ assertEquals(0, paths.getScanServer(ResourceGroupPredicate.exact(ResourceGroupId.of("TTEST")),
+ AddressSelector.all(), true).size());
getCluster().getClusterControl().stopAllServers(ServerType.COMPACTOR);
- Wait.waitFor(() -> paths.getCompactor(rg -> true, AddressSelector.all(), true).size() == 0);
+ Wait.waitFor(
+ () -> paths.getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size()
+ == 0);
getCluster().getClusterControl().stopAllServers(ServerType.SCAN_SERVER);
- Wait.waitFor(() -> paths.getScanServer(rg -> true, AddressSelector.all(), true).size() == 0);
+ Wait.waitFor(
+ () -> paths.getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size()
+ == 0);
getCluster().getClusterControl().stopAllServers(ServerType.GARBAGE_COLLECTOR);
@@ -125,8 +115,12 @@ public class ServiceLockPathsIT extends AccumuloClusterHarness {
getCluster().getClusterControl().stopAllServers(ServerType.TABLET_SERVER);
- Wait.waitFor(() -> paths.getTabletServer(rg -> true, AddressSelector.all(), true).size() == 0);
- Wait.waitFor(() -> paths.getTabletServer(rg -> true, AddressSelector.all(), false).size() == 2);
+ Wait.waitFor(
+ () -> paths.getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size()
+ == 0);
+ Wait.waitFor(
+ () -> paths.getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), false).size()
+ == 2);
}
diff --git a/test/src/main/java/org/apache/accumulo/test/manager/SuspendedTabletsIT.java b/test/src/main/java/org/apache/accumulo/test/manager/SuspendedTabletsIT.java
index 033fab023c..5fc0563322 100644
--- a/test/src/main/java/org/apache/accumulo/test/manager/SuspendedTabletsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/manager/SuspendedTabletsIT.java
@@ -54,6 +54,7 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.metadata.schema.Ample;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType;
@@ -310,7 +311,7 @@ public class SuspendedTabletsIT extends AccumuloClusterHarness {
ThriftClientTypes.MANAGER.executeVoid(ctx, client -> {
log.info("Sending shutdown command to {} via ManagerClientService", ts);
client.shutdownTabletServer(null, ctx.rpcCreds(), ts, false);
- });
+ }, ResourceGroupPredicate.DEFAULT_RG_ONLY);
} catch (AccumuloSecurityException | AccumuloException e) {
throw new RuntimeException("Error calling shutdownTabletServer for " + ts, e);
}
diff --git a/test/src/main/java/org/apache/accumulo/test/upgrade/UpgradeIT.java b/test/src/main/java/org/apache/accumulo/test/upgrade/UpgradeIT.java
index e0c72ca2c2..ae1935564f 100644
--- a/test/src/main/java/org/apache/accumulo/test/upgrade/UpgradeIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/upgrade/UpgradeIT.java
@@ -35,6 +35,7 @@ import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.harness.AccumuloClusterHarness;
import org.apache.accumulo.manager.Manager;
@@ -115,26 +116,26 @@ public class UpgradeIT extends AccumuloClusterHarness {
// Confirm all servers down
Wait.waitFor(() -> getServerContext().getServerPaths()
- .getCompactor((rg) -> true, AddressSelector.all(), true).size() == 0);
+ .getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 0);
Wait.waitFor(() -> getServerContext().getServerPaths().getGarbageCollector(true) == null);
Wait.waitFor(() -> getServerContext().getServerPaths().getManager(true) == null);
Wait.waitFor(() -> getServerContext().getServerPaths()
- .getScanServer((rg) -> true, AddressSelector.all(), true).size() == 0);
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 0);
Wait.waitFor(() -> getServerContext().getServerPaths()
- .getTabletServer((rg) -> true, AddressSelector.all(), true).size() == 0);
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 0);
assertThrows(IllegalStateException.class,
() -> assertTimeoutPreemptively(Duration.ofMinutes(2), () -> getCluster().start()));
// Confirm no servers started
Wait.waitFor(() -> getServerContext().getServerPaths()
- .getCompactor((rg) -> true, AddressSelector.all(), true).size() == 0);
+ .getCompactor(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 0);
Wait.waitFor(() -> getServerContext().getServerPaths().getGarbageCollector(true) == null);
Wait.waitFor(() -> getServerContext().getServerPaths().getManager(true) == null);
Wait.waitFor(() -> getServerContext().getServerPaths()
- .getScanServer((rg) -> true, AddressSelector.all(), true).size() == 0);
+ .getScanServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 0);
Wait.waitFor(() -> getServerContext().getServerPaths()
- .getTabletServer((rg) -> true, AddressSelector.all(), true).size() == 0);
+ .getTabletServer(ResourceGroupPredicate.ANY, AddressSelector.all(), true).size() == 0);
// Validate the exception from the servers
List<String> args = new ArrayList<>();