This is an automated email from the ASF dual-hosted git repository.
kturner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new a0d68172dc Moves set of shutting down tservers from manager memory to
ZK (#6224)
a0d68172dc is described below
commit a0d68172dc00268e67a91166761e8407c1c1a43f
Author: Keith Turner <[email protected]>
AuthorDate: Thu Apr 2 14:57:14 2026 -0700
Moves set of shutting down tservers from manager memory to ZK (#6224)
The set of shutting down tservers was causing system fate operations to
have to run on the primary manager because this was an in memory set.
This caused fate to have different code paths to user vs system fate,
this in turn caused problems when trying to distribute compaction
coordination.
To fix this problem moved the set from an in memory set to a set in
zookeeper. The set is managed by fate operations which simplifies the
existing code. Only fate operations add and remove from the set and fate
keys are used to ensure only one fate operation runs at a time for a
tserver instance. The previous in memory set had a lot of code to try to
keep
it in sync with reality, that is all gone now. There were many bugs with
this code in the past.
After this change is made fate can be simplified in a follow on commit
to remove all specialization for the primary manager. Also the monitor
can now directly access this set instead of making an RPC to the
manager, will open a follow on issue for this.
---
.../java/org/apache/accumulo/core/Constants.java | 2 +
.../org/apache/accumulo/core/fate/FateKey.java | 43 ++++++++-
.../accumulo/server/init/ZooKeeperInitializer.java | 2 +
.../java/org/apache/accumulo/manager/Manager.java | 76 +++++++---------
.../manager/ManagerClientServiceHandler.java | 39 ++++----
.../manager/tserverOps/BeginTserverShutdown.java | 80 +++++++++++++++++
.../manager/tserverOps/ShutdownTServer.java | 75 +++++++++-------
.../accumulo/manager/upgrade/Upgrader11to12.java | 11 +++
.../manager/tableOps/ShutdownTServerTest.java | 100 ---------------------
9 files changed, 228 insertions(+), 200 deletions(-)
diff --git a/core/src/main/java/org/apache/accumulo/core/Constants.java
b/core/src/main/java/org/apache/accumulo/core/Constants.java
index eb8ba1059e..d2c556f2c2 100644
--- a/core/src/main/java/org/apache/accumulo/core/Constants.java
+++ b/core/src/main/java/org/apache/accumulo/core/Constants.java
@@ -74,6 +74,8 @@ public class Constants {
public static final String ZDEAD = "/dead";
public static final String ZDEADTSERVERS = ZDEAD + "/tservers";
+ public static final String ZSHUTTING_DOWN_TSERVERS =
"/shutting-down-tservers";
+
public static final String ZFATE = "/fate";
public static final String ZNEXT_FILE = "/next_file";
diff --git a/core/src/main/java/org/apache/accumulo/core/fate/FateKey.java
b/core/src/main/java/org/apache/accumulo/core/fate/FateKey.java
index 650dca01af..19641c000a 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/FateKey.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/FateKey.java
@@ -27,6 +27,7 @@ import java.util.Objects;
import java.util.Optional;
import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.TServerInstance;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
@@ -41,12 +42,14 @@ public class FateKey {
private final FateKeyType type;
private final Optional<KeyExtent> keyExtent;
private final Optional<ExternalCompactionId> compactionId;
+ private final Optional<TServerInstance> tServerInstance;
private final byte[] serialized;
private FateKey(FateKeyType type, KeyExtent keyExtent) {
this.type = Objects.requireNonNull(type);
this.keyExtent = Optional.of(keyExtent);
this.compactionId = Optional.empty();
+ this.tServerInstance = Optional.empty();
this.serialized = serialize(type, keyExtent);
}
@@ -54,15 +57,25 @@ public class FateKey {
this.type = Objects.requireNonNull(type);
this.keyExtent = Optional.empty();
this.compactionId = Optional.of(compactionId);
+ this.tServerInstance = Optional.empty();
this.serialized = serialize(type, compactionId);
}
+ private FateKey(FateKeyType type, TServerInstance tServerInstance) {
+ this.type = Objects.requireNonNull(type);
+ this.keyExtent = Optional.empty();
+ this.compactionId = Optional.empty();
+ this.tServerInstance = Optional.of(tServerInstance);
+ this.serialized = serialize(type, tServerInstance);
+ }
+
private FateKey(byte[] serialized) {
try (DataInputBuffer buffer = new DataInputBuffer()) {
buffer.reset(serialized, serialized.length);
this.type = FateKeyType.valueOf(buffer.readUTF());
this.keyExtent = deserializeKeyExtent(type, buffer);
this.compactionId = deserializeCompactionId(type, buffer);
+ this.tServerInstance = deserializeTserverId(type, buffer);
this.serialized = serialized;
} catch (IOException e) {
throw new UncheckedIOException(e);
@@ -127,8 +140,12 @@ public class FateKey {
return new FateKey(FateKeyType.MERGE, extent);
}
+ public static FateKey forShutdown(TServerInstance tServerInstance) {
+ return new FateKey(FateKeyType.TSERVER_SHUTDOWN, tServerInstance);
+ }
+
public enum FateKeyType {
- SPLIT, COMPACTION_COMMIT, MERGE
+ SPLIT, COMPACTION_COMMIT, MERGE, TSERVER_SHUTDOWN
}
private static byte[] serialize(FateKeyType type, KeyExtent ke) {
@@ -155,22 +172,42 @@ public class FateKey {
}
}
+ private static byte[] serialize(FateKeyType type, TServerInstance
tServerInstance) {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(baos)) {
+ dos.writeUTF(type.toString());
+ dos.writeUTF(tServerInstance.getHostPortSession());
+ dos.close();
+ return baos.toByteArray();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
private static Optional<KeyExtent> deserializeKeyExtent(FateKeyType type,
DataInputBuffer buffer)
throws IOException {
return switch (type) {
case SPLIT, MERGE -> Optional.of(KeyExtent.readFrom(buffer));
- case COMPACTION_COMMIT -> Optional.empty();
+ case COMPACTION_COMMIT, TSERVER_SHUTDOWN -> Optional.empty();
};
}
private static Optional<ExternalCompactionId>
deserializeCompactionId(FateKeyType type,
DataInputBuffer buffer) throws IOException {
return switch (type) {
- case SPLIT, MERGE -> Optional.empty();
+ case SPLIT, MERGE, TSERVER_SHUTDOWN -> Optional.empty();
case COMPACTION_COMMIT ->
Optional.of(ExternalCompactionId.of(buffer.readUTF()));
};
}
+ private static Optional<TServerInstance> deserializeTserverId(FateKeyType
type,
+ DataInputBuffer buffer) throws IOException {
+ return switch (type) {
+ case SPLIT, MERGE, COMPACTION_COMMIT -> Optional.empty();
+ case TSERVER_SHUTDOWN -> Optional.of(new
TServerInstance(buffer.readUTF()));
+ };
+ }
+
@Override
public String toString() {
var buf = new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE);
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java
b/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java
index 464a6dae5a..b23c674680 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java
@@ -178,6 +178,8 @@ public class ZooKeeperInitializer {
ZooUtil.NodeExistsPolicy.FAIL);
zrwChroot.putPersistentData(Constants.ZMANAGER_ASSISTANT_LOCK,
EMPTY_BYTE_ARRAY,
ZooUtil.NodeExistsPolicy.FAIL);
+ zrwChroot.putPersistentData(Constants.ZSHUTTING_DOWN_TSERVERS,
EMPTY_BYTE_ARRAY,
+ ZooUtil.NodeExistsPolicy.FAIL);
}
/**
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 60e09f3dd8..b1edc70857 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
@@ -58,6 +58,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.Predicate;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.cli.ServerOpts;
@@ -205,7 +206,6 @@ public class Manager extends AbstractServer
private final List<TabletGroupWatcher> watchers = new ArrayList<>();
final Map<TServerInstance,AtomicInteger> badServers =
Collections.synchronizedMap(new HashMap<>());
- final Set<TServerInstance> serversToShutdown =
Collections.synchronizedSet(new HashSet<>());
final EventCoordinator nextEvent = new EventCoordinator();
RecoveryManager recoveryManager = null;
private final ManagerTime timeKeeper;
@@ -556,10 +556,6 @@ public class Manager extends AbstractServer
aconf.getTimeInMillis(Property.MANAGER_RECOVERY_WAL_EXISTENCE_CACHE_TIME);
}
- public TServerConnection getConnection(TServerInstance server) {
- return tserverSet.getConnection(server);
- }
-
void setManagerGoalState(ManagerGoalState state) {
try {
getContext().getZooSession().asReaderWriter().putPersistentData(Constants.ZMANAGER_GOAL_STATE,
@@ -661,6 +657,7 @@ public class Manager extends AbstractServer
}
boolean canBalance(DataLevel dataLevel, TServerStatus tServerStatus) {
+ Set<TServerInstance> serversToShutdown;
if (!badServers.isEmpty()) {
log.debug("not balancing {} because the balance information is
out-of-date {}", dataLevel,
badServers.keySet());
@@ -668,7 +665,7 @@ public class Manager extends AbstractServer
} else if (getManagerGoalState() == ManagerGoalState.CLEAN_STOP) {
log.debug("not balancing {} because the manager is attempting to stop
cleanly", dataLevel);
return false;
- } else if (!serversToShutdown.isEmpty()) {
+ } else if (!(serversToShutdown = shutdownServers()).isEmpty()) {
log.debug("not balancing {} while shutting down servers {}", dataLevel,
serversToShutdown);
return false;
} else {
@@ -760,7 +757,6 @@ public class Manager extends AbstractServer
log.debug("stopping {} tablet servers",
currentServers.size());
for (TServerInstance server : currentServers) {
try {
- serversToShutdown.add(server);
tserverSet.getConnection(server).fastHalt(primaryManagerLock);
} catch (TException e) {
// its probably down, and we don't care
@@ -1585,38 +1581,35 @@ public class Manager extends AbstractServer
obit.delete(up.getHostPort());
}
}
- for (TServerInstance dead : deleted) {
- String cause = "unexpected failure";
- if (serversToShutdown.contains(dead)) {
- cause = "clean shutdown"; // maybe an incorrect assumption
+
+ if (!deleted.isEmpty()) {
+ // This set is read from zookeeper, so only get it if its actually
needed
+ var serversToShutdown = shutdownServers();
+ for (TServerInstance dead : deleted) {
+ String cause = "unexpected failure";
+ if (serversToShutdown.contains(dead)) {
+ cause = "clean shutdown"; // maybe an incorrect assumption
+ }
+ if (!getManagerGoalState().equals(ManagerGoalState.CLEAN_STOP)) {
+ obit.post(dead.getHostPort(), cause);
+ }
}
- if (!getManagerGoalState().equals(ManagerGoalState.CLEAN_STOP)) {
- obit.post(dead.getHostPort(), cause);
+
+ Set<TServerInstance> unexpected = new HashSet<>(deleted);
+ unexpected.removeAll(serversToShutdown);
+ if (!unexpected.isEmpty()
+ && (stillManager() &&
!getManagerGoalState().equals(ManagerGoalState.CLEAN_STOP))) {
+ log.warn("Lost servers {}", unexpected);
}
+ badServers.keySet().removeAll(deleted);
}
- Set<TServerInstance> unexpected = new HashSet<>(deleted);
- unexpected.removeAll(this.serversToShutdown);
- if (!unexpected.isEmpty()
- && (stillManager() &&
!getManagerGoalState().equals(ManagerGoalState.CLEAN_STOP))) {
- log.warn("Lost servers {}", unexpected);
- }
- serversToShutdown.removeAll(deleted);
- badServers.keySet().removeAll(deleted);
// clear out any bad server with the same host/port as a new server
synchronized (badServers) {
cleanListByHostAndPort(badServers.keySet(), deleted, added);
}
- synchronized (serversToShutdown) {
- cleanListByHostAndPort(serversToShutdown, deleted, added);
- }
nextEvent.event("There are now %d tablet servers", current.size());
}
-
- // clear out any servers that are no longer current
- // this is needed when we are using a fate operation to shutdown a tserver
as it
- // will continue to add the server to the serversToShutdown (ACCUMULO-4410)
- serversToShutdown.retainAll(current.getCurrentServers());
}
static void cleanListByHostAndPort(Collection<TServerInstance> badServers,
@@ -1663,15 +1656,6 @@ public class Manager extends AbstractServer
return tserverSet.getSnapshot();
}
- // recovers state from the persistent transaction to shutdown a server
- public boolean shutdownTServer(TServerInstance server) {
- if (serversToShutdown.add(server)) {
- nextEvent.event("Tablet Server shutdown requested for %s", server);
- return true;
- }
- return false;
- }
-
public EventCoordinator getEventCoordinator() {
return nextEvent;
}
@@ -1719,12 +1703,8 @@ public class Manager extends AbstractServer
result.state = getManagerState();
result.goalState = getManagerGoalState();
result.unassignedTablets = displayUnassigned();
- result.serversShuttingDown = new HashSet<>();
- synchronized (serversToShutdown) {
- for (TServerInstance server : serversToShutdown) {
- result.serversShuttingDown.add(server.getHostPort());
- }
- }
+ result.serversShuttingDown =
+
shutdownServers().stream().map(TServerInstance::getHostPort).collect(Collectors.toSet());
DeadServerList obit = new DeadServerList(getContext());
result.deadTabletServers = obit.getList();
return result;
@@ -1738,8 +1718,12 @@ public class Manager extends AbstractServer
}
public Set<TServerInstance> shutdownServers() {
- synchronized (serversToShutdown) {
- return Set.copyOf(serversToShutdown);
+ try {
+ List<String> children =
+
getContext().getZooSession().asReader().getChildren(Constants.ZSHUTTING_DOWN_TSERVERS);
+ return
children.stream().map(TServerInstance::new).collect(Collectors.toSet());
+ } catch (KeeperException | InterruptedException e) {
+ throw new RuntimeException(e);
}
}
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/ManagerClientServiceHandler.java
b/server/manager/src/main/java/org/apache/accumulo/manager/ManagerClientServiceHandler.java
index f96332dc9d..d88a94516d 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/ManagerClientServiceHandler.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/ManagerClientServiceHandler.java
@@ -33,7 +33,9 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.Constants;
@@ -62,6 +64,7 @@ import org.apache.accumulo.core.fate.Fate;
import org.apache.accumulo.core.fate.FateClient;
import org.apache.accumulo.core.fate.FateId;
import org.apache.accumulo.core.fate.FateInstanceType;
+import org.apache.accumulo.core.fate.FateKey;
import org.apache.accumulo.core.fate.TraceRepo;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.manager.state.tables.TableState;
@@ -85,7 +88,7 @@ import
org.apache.accumulo.core.securityImpl.thrift.TDelegationToken;
import org.apache.accumulo.core.securityImpl.thrift.TDelegationTokenConfig;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.manager.tableOps.FateEnv;
-import org.apache.accumulo.manager.tserverOps.ShutdownTServer;
+import org.apache.accumulo.manager.tserverOps.BeginTserverShutdown;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.client.ClientServiceHandler;
import org.apache.accumulo.server.conf.store.NamespacePropKey;
@@ -336,16 +339,16 @@ public class ManagerClientServiceHandler implements
ManagerClientService.Iface {
}
FateClient<FateEnv> fate = manager.fateClient(FateInstanceType.META);
- FateId fateId = fate.startTransaction();
- String msg = "Shutdown tserver " + tabletServer;
+ var repo = new TraceRepo<>(
+ new BeginTserverShutdown(doomed,
manager.tserverSet.getResourceGroup(doomed), force));
- fate.seedTransaction(Fate.FateOperation.SHUTDOWN_TSERVER, fateId,
- new TraceRepo<>(
- new ShutdownTServer(doomed,
manager.tserverSet.getResourceGroup(doomed), force)),
- false, msg);
- fate.waitForCompletion(fateId);
- fate.delete(fateId);
+ CompletableFuture<Optional<FateId>> future;
+ try (var seeder = fate.beginSeeding()) {
+ future =
seeder.attemptToSeedTransaction(Fate.FateOperation.SHUTDOWN_TSERVER,
+ FateKey.forShutdown(doomed), repo, true);
+ }
+ future.join().ifPresent(fate::waitForCompletion);
log.debug("FATE op shutting down " + tabletServer + " finished");
}
@@ -360,17 +363,15 @@ public class ManagerClientServiceHandler implements
ManagerClientService.Iface {
}
log.info("Tablet Server {} has reported it's shutting down", tabletServer);
var tserver = new TServerInstance(tabletServer);
- if (manager.shutdownTServer(tserver)) {
- // If there is an exception seeding the fate tx this should cause the
RPC to fail which should
- // cause the tserver to halt. Because of that not making an attempt to
handle failure here.
- FateClient<FateEnv> fate = manager.fateClient(FateInstanceType.META);
- var tid = fate.startTransaction();
- String msg = "Shutdown tserver " + tabletServer;
+ // If there is an exception seeding the fate tx this should cause the RPC
to fail which should
+ // cause the tserver to halt. Because of that not making an attempt to
handle failure here.
+ FateClient<FateEnv> fate = manager.fateClient(FateInstanceType.META);
- fate.seedTransaction(Fate.FateOperation.SHUTDOWN_TSERVER, tid,
- new TraceRepo<>(new ShutdownTServer(tserver,
ResourceGroupId.of(resourceGroup), false)),
- true, msg);
- }
+ var repo = new TraceRepo<>(
+ new BeginTserverShutdown(tserver, ResourceGroupId.of(resourceGroup),
false));
+ // only seed a new transaction if nothing is running for this tserver
+ fate.seedTransaction(Fate.FateOperation.SHUTDOWN_TSERVER,
FateKey.forShutdown(tserver), repo,
+ true);
}
@Override
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/BeginTserverShutdown.java
b/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/BeginTserverShutdown.java
new file mode 100644
index 0000000000..358595ebca
--- /dev/null
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/BeginTserverShutdown.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.manager.tserverOps;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.data.ResourceGroupId;
+import org.apache.accumulo.core.fate.FateId;
+import org.apache.accumulo.core.fate.Repo;
+import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
+import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.manager.tableOps.AbstractFateOperation;
+import org.apache.accumulo.manager.tableOps.FateEnv;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.net.HostAndPort;
+
+public class BeginTserverShutdown extends AbstractFateOperation {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger log =
LoggerFactory.getLogger(BeginTserverShutdown.class);
+
+ private final ResourceGroupId resourceGroup;
+ private final HostAndPort hostAndPort;
+ private final String serverSession;
+ private final boolean force;
+
+ public BeginTserverShutdown(TServerInstance server, ResourceGroupId
resourceGroup,
+ boolean force) {
+ this.hostAndPort = server.getHostAndPort();
+ this.resourceGroup = resourceGroup;
+ this.serverSession = server.getSession();
+ this.force = force;
+ }
+
+ static String createPath(HostAndPort hostPort, String session) {
+ return Constants.ZSHUTTING_DOWN_TSERVERS + "/"
+ + new TServerInstance(hostPort, session).getHostPortSession();
+ }
+
+ @Override
+ public Repo<FateEnv> call(FateId fateId, FateEnv environment) throws
Exception {
+ if (!force) {
+ String path = createPath(hostAndPort, serverSession);
+ // Because these fate operation are seeded with a fate key there should
only ever be one fate
+ // operation running for a tserver instance, so do not need to worry
about race conditions
+ // here
+
environment.getContext().getZooSession().asReaderWriter().putPersistentData(path,
new byte[0],
+ ZooUtil.NodeExistsPolicy.SKIP);
+ log.trace("{} created {}", fateId, path);
+ }
+ return new ShutdownTServer(hostAndPort, serverSession, resourceGroup,
force);
+ }
+
+ @Override
+ public void undo(FateId fateId, FateEnv environment) throws Exception {
+ if (!force) {
+ String path = createPath(hostAndPort, serverSession);
+ environment.getContext().getZooSession().asReaderWriter().delete(path);
+ log.trace("{} removed {}", fateId, path);
+ }
+ }
+}
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java
b/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java
index 147b8206cc..eb456a8a3c 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/tserverOps/ShutdownTServer.java
@@ -19,6 +19,7 @@
package org.apache.accumulo.manager.tserverOps;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static
org.apache.accumulo.manager.tserverOps.BeginTserverShutdown.createPath;
import org.apache.accumulo.core.data.ResourceGroupId;
import org.apache.accumulo.core.fate.FateId;
@@ -28,10 +29,12 @@ import
org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.lock.ServiceLock;
import org.apache.accumulo.core.manager.thrift.TabletServerStatus;
import org.apache.accumulo.core.metadata.TServerInstance;
-import org.apache.accumulo.manager.Manager;
+import org.apache.accumulo.core.rpc.ThriftUtil;
+import org.apache.accumulo.core.rpc.clients.ThriftClientTypes;
+import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService;
+import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.manager.tableOps.AbstractFateOperation;
import org.apache.accumulo.manager.tableOps.FateEnv;
-import org.apache.accumulo.server.manager.LiveTServerSet.TServerConnection;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,10 +50,11 @@ public class ShutdownTServer extends AbstractFateOperation {
private final String serverSession;
private final boolean force;
- public ShutdownTServer(TServerInstance server, ResourceGroupId
resourceGroup, boolean force) {
- this.hostAndPort = server.getHostAndPort();
+ public ShutdownTServer(HostAndPort hostAndPort, String serverSession,
+ ResourceGroupId resourceGroup, boolean force) {
+ this.hostAndPort = hostAndPort;
this.resourceGroup = resourceGroup;
- this.serverSession = server.getSession();
+ this.serverSession = serverSession;
this.force = force;
}
@@ -62,35 +66,38 @@ public class ShutdownTServer extends AbstractFateOperation {
return 0;
}
- // Inform the manager that we want this server to shutdown
- Manager manager = (Manager) env;
- manager.shutdownTServer(server);
-
- if (manager.onlineTabletServers().contains(server)) {
- TServerConnection connection = manager.getConnection(server);
- if (connection != null) {
- try {
- TabletServerStatus status = connection.getTableMap(false);
- if (status.tableMap != null && status.tableMap.isEmpty()) {
- log.info("tablet server hosts no tablets {}", server);
- connection.halt(manager.getServiceLock());
- log.info("tablet server asked to halt {}", server);
- return 0;
- } else {
- log.info("tablet server {} still has tablets for tables: {}",
server,
- (status.tableMap == null) ? "null" : status.tableMap.keySet());
- }
- } catch (TTransportException ex) {
- // expected
- } catch (Exception ex) {
- log.error("Error talking to tablet server {}: ", server, ex);
- }
+ if (env.onlineTabletServers().contains(server)) {
+
+ TabletServerClientService.Client client = null;
- // If the connection was non-null and we could communicate with it
- // give the manager some more time to tell it to stop and for the
- // tserver to ack the request and stop itself.
- return 1000;
+ try {
+ client =
+ ThriftUtil.getClient(ThriftClientTypes.TABLET_SERVER, hostAndPort,
env.getContext());
+ TabletServerStatus status =
+ client.getTabletServerStatus(TraceUtil.traceInfo(),
env.getContext().rpcCreds());
+ if (status.tableMap != null && status.tableMap.isEmpty()) {
+ log.info("tablet server hosts no tablets {}", server);
+ client.halt(TraceUtil.traceInfo(), env.getContext().rpcCreds(),
+ env.getServiceLock().getLockID().serialize());
+ log.info("tablet server asked to halt {}", server);
+ return 0;
+ } else {
+ log.info("tablet server {} still has tablets for tables: {}", server,
+ (status.tableMap == null) ? "null" : status.tableMap.keySet());
+ }
+ } catch (TTransportException ex) {
+ // expected
+ } catch (Exception ex) {
+ log.error("Error talking to tablet server {}: ", server, ex);
+ } finally {
+ ThriftUtil.returnClient(client, env.getContext());
}
+
+ // If the connection was non-null and we could communicate with it
+ // give the manager some more time to tell it to stop and for the
+ // tserver to ack the request and stop itself.
+ return 1000;
+
}
return 0;
@@ -108,6 +115,10 @@ public class ShutdownTServer extends AbstractFateOperation
{
env.getContext().getServerPaths().createDeadTabletServerPath(resourceGroup,
hostAndPort);
zoo.putPersistentData(path.toString(), "forced down".getBytes(UTF_8),
NodeExistsPolicy.OVERWRITE);
+ } else {
+ String path = createPath(hostAndPort, serverSession);
+ env.getContext().getZooSession().asReaderWriter().delete(path);
+ log.trace("{} removed {}", fateId, path);
}
return null;
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
b/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
index a230dc7149..f88a9d59c4 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java
@@ -255,6 +255,8 @@ public class Upgrader11to12 implements Upgrader {
moveTableProperties(context);
LOG.info("Add assistant manager node");
addAssistantManager(context);
+ LOG.info("Adding shutting down tservers node");
+ addShuttingDownTservers(context);
}
@Override
@@ -308,6 +310,15 @@ public class Upgrader11to12 implements Upgrader {
}
}
+ private static void addShuttingDownTservers(ServerContext context) {
+ try {
+
context.getZooSession().asReaderWriter().putPersistentData(Constants.ZSHUTTING_DOWN_TSERVERS,
+ new byte[0], ZooUtil.NodeExistsPolicy.SKIP);
+ } catch (KeeperException | InterruptedException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
private static void addCompactionsNode(ServerContext context) {
try {
context.getZooSession().asReaderWriter().putPersistentData(Constants.ZCOMPACTIONS,
diff --git
a/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/ShutdownTServerTest.java
b/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/ShutdownTServerTest.java
deleted file mode 100644
index 8dbe53bf12..0000000000
---
a/server/manager/src/test/java/org/apache/accumulo/manager/tableOps/ShutdownTServerTest.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.accumulo.manager.tableOps;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.UUID;
-
-import org.apache.accumulo.core.data.ResourceGroupId;
-import org.apache.accumulo.core.fate.FateId;
-import org.apache.accumulo.core.fate.FateInstanceType;
-import org.apache.accumulo.core.manager.thrift.TableInfo;
-import org.apache.accumulo.core.manager.thrift.TabletServerStatus;
-import org.apache.accumulo.core.metadata.TServerInstance;
-import org.apache.accumulo.manager.Manager;
-import org.apache.accumulo.manager.tserverOps.ShutdownTServer;
-import org.apache.accumulo.server.manager.LiveTServerSet.TServerConnection;
-import org.easymock.EasyMock;
-import org.junit.jupiter.api.Test;
-
-import com.google.common.net.HostAndPort;
-
-public class ShutdownTServerTest {
-
- @Test
- public void testSingleShutdown() throws Exception {
- HostAndPort hap = HostAndPort.fromParts("localhost", 1234);
- final TServerInstance tserver = new TServerInstance(hap, "fake");
- final boolean force = false;
-
- final ShutdownTServer op = new ShutdownTServer(tserver,
ResourceGroupId.DEFAULT, force);
-
- final Manager manager = EasyMock.createMock(Manager.class);
- final FateId fateId = FateId.from(FateInstanceType.USER,
UUID.randomUUID());
-
- final TServerConnection tserverCnxn =
EasyMock.createMock(TServerConnection.class);
- final TabletServerStatus status = new TabletServerStatus();
- status.tableMap = new HashMap<>();
- // Put in a table info record, don't care what
- status.tableMap.put("a_table", new TableInfo());
-
- EasyMock.expect(manager.shutdownTServer(tserver)).andReturn(true).once();
-
EasyMock.expect(manager.onlineTabletServers()).andReturn(Collections.singleton(tserver));
- EasyMock.expect(manager.getConnection(tserver)).andReturn(tserverCnxn);
- EasyMock.expect(tserverCnxn.getTableMap(false)).andReturn(status);
-
- EasyMock.replay(tserverCnxn, manager);
-
- // FATE op is not ready
- long wait = op.isReady(fateId, manager);
- assertTrue(wait > 0, "Expected wait to be greater than 0");
-
- EasyMock.verify(tserverCnxn, manager);
-
- // Reset the mocks
- EasyMock.reset(tserverCnxn, manager);
-
- // reset the table map to the empty set to simulate all tablets unloaded
- status.tableMap = new HashMap<>();
- EasyMock.expect(manager.shutdownTServer(tserver)).andReturn(false).once();
-
EasyMock.expect(manager.onlineTabletServers()).andReturn(Collections.singleton(tserver));
- EasyMock.expect(manager.getConnection(tserver)).andReturn(tserverCnxn);
- EasyMock.expect(tserverCnxn.getTableMap(false)).andReturn(status);
- EasyMock.expect(manager.getServiceLock()).andReturn(null);
- tserverCnxn.halt(null);
- EasyMock.expectLastCall().once();
-
- EasyMock.replay(tserverCnxn, manager);
-
- // FATE op is not ready
- wait = op.isReady(fateId, manager);
- assertEquals(0, wait, "Expected wait to be 0");
-
- var op2 = op.call(fateId, manager);
- assertNull(op2, "Expected no follow on step");
-
- EasyMock.verify(tserverCnxn, manager);
- }
-
-}
