This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch elasticity
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/elasticity by this push:
new 5d12597b4d Add LogSorter to Compactor and ScanServer (#4239)
5d12597b4d is described below
commit 5d12597b4dedb60524775d0411405bafc3ee86ce
Author: Dave Marion <dlmar...@apache.org>
AuthorDate: Wed Feb 14 08:06:49 2024 -0500
Add LogSorter to Compactor and ScanServer (#4239)
Adds LogSorter to ScanServer and Compactor such that
all available processes will participate when a failure
occurs that leaves a Tablet with walogs. Modified LogSorter
and DistributedWorkQueue so that the Compactor could call
the LogSorter and have its tasks execute serially in the
Compactor thread.
Fixes #4232
---
.../org/apache/accumulo/core/conf/Property.java | 3 +
.../MiniAccumuloClusterControl.java | 4 +
.../server/zookeeper/DistributedWorkQueue.java | 97 ++++++---
server/compactor/pom.xml | 4 +
.../org/apache/accumulo/compactor/Compactor.java | 11 +
.../org/apache/accumulo/tserver/ScanServer.java | 17 ++
.../org/apache/accumulo/tserver/TabletServer.java | 20 +-
.../org/apache/accumulo/tserver/log/LogSorter.java | 25 ++-
.../java/org/apache/accumulo/test/RecoveryIT.java | 231 +++++++++++++++++++++
9 files changed, 371 insertions(+), 41 deletions(-)
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index 12f69d2ff9..22c2606c85 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -520,6 +520,9 @@ public enum Property {
@Experimental
SSERV_THREADCHECK("sserver.server.threadcheck.time", "1s",
PropertyType.TIMEDURATION,
"The time between adjustments of the thrift server thread pool.",
"2.1.0"),
+ @Experimental
+ SSERV_WAL_SORT_MAX_CONCURRENT("sserver.wal.sort.concurrent.max", "2",
PropertyType.COUNT,
+ "The maximum number of threads to use to sort logs during recovery.",
"4.0.0"),
// properties that are specific to tablet server behavior
TSERV_PREFIX("tserver.", null, PropertyType.PREFIX,
"Properties in this category affect the behavior of the tablet
servers.", "1.3.5"),
diff --git
a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
index b3977dab00..21e831c3e0 100644
---
a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
+++
b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterControl.java
@@ -519,4 +519,8 @@ public class MiniAccumuloClusterControl implements
ClusterControl {
stop(server, hostname);
}
+ public List<Process> getTabletServers(String resourceGroup) {
+ return tabletServerProcesses.get(resourceGroup);
+ }
+
}
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
index cd667909d0..2263e02af4 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
@@ -27,6 +27,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -57,7 +58,6 @@ public class DistributedWorkQueue {
private static final Logger log =
LoggerFactory.getLogger(DistributedWorkQueue.class);
- private ThreadPoolExecutor threadPool;
private ZooReaderWriter zoo;
private String path;
private ServerContext context;
@@ -65,12 +65,20 @@ public class DistributedWorkQueue {
private AtomicInteger numTask = new AtomicInteger(0);
- private void lookForWork(final Processor processor, List<String> children) {
+ /**
+ * Finds a child in {@code children} that is not currently being processed
and adds a Runnable to
+ * the {@code executor} that invokes the {@code processor}. The Runnable
will recursively call
+ * {@code lookForWork} after it invokes the {@code processor} such that it
will continue to look
+ * for children that need work until that condition is exhausted. This
method will return early if
+ * the number of currently running tasks is larger than {@code maxThreads}.
+ */
+ private void lookForWork(final Processor processor, final List<String>
children,
+ final ExecutorService executor, final int maxThreads) {
if (children.isEmpty()) {
return;
}
- if (numTask.get() >= threadPool.getCorePoolSize()) {
+ if (numTask.get() >= maxThreads) {
return;
}
@@ -102,7 +110,7 @@ public class DistributedWorkQueue {
}
// Great... we got the lock, but maybe we're too busy
- if (numTask.get() >= threadPool.getCorePoolSize()) {
+ if (numTask.get() >= maxThreads) {
zoo.recursiveDelete(lockPath, NodeMissingPolicy.SKIP);
break;
}
@@ -143,7 +151,7 @@ public class DistributedWorkQueue {
try {
// its important that this is called after numTask is decremented
- lookForWork(processor, zoo.getChildren(path));
+ lookForWork(processor, zoo.getChildren(path), executor,
maxThreads);
} catch (KeeperException e) {
log.error("Failed to look for work", e);
} catch (InterruptedException e) {
@@ -153,7 +161,7 @@ public class DistributedWorkQueue {
};
numTask.incrementAndGet();
- threadPool.execute(task);
+ executor.execute(task);
}
} catch (Exception t) {
@@ -186,40 +194,62 @@ public class DistributedWorkQueue {
return context;
}
- public void startProcessing(final Processor processor, ThreadPoolExecutor
executorService)
- throws KeeperException, InterruptedException {
+ public long getCheckInterval() {
+ return this.timerPeriod;
+ }
- threadPool = executorService;
+ /**
+ * Finds the children at the path passed in the constructor and calls {@code
lookForWork} which
+ * will attempt to process all of the currently available work
+ */
+ public void processExistingWork(final Processor processor, ExecutorService
executor,
+ final int maxThreads, boolean setWatch) throws KeeperException,
InterruptedException {
zoo.mkdirs(path);
zoo.mkdirs(path + "/" + LOCKS_NODE);
- List<String> children = zoo.getChildren(path, new Watcher() {
- @Override
- public void process(WatchedEvent event) {
- switch (event.getType()) {
- case NodeChildrenChanged:
- if (event.getPath().equals(path)) {
- try {
- lookForWork(processor, zoo.getChildren(path, this));
- } catch (KeeperException e) {
- log.error("Failed to look for work at path {}; {}", path,
event, e);
- } catch (InterruptedException e) {
- log.info("Interrupted looking for work at path {}; {}", path,
event, e);
+ List<String> children = null;
+ if (setWatch) {
+ children = zoo.getChildren(path, new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ switch (event.getType()) {
+ case NodeChildrenChanged:
+ if (event.getPath().equals(path)) {
+ try {
+ lookForWork(processor, zoo.getChildren(path, this),
executor, maxThreads);
+ } catch (KeeperException e) {
+ log.error("Failed to look for work at path {}; {}", path,
event, e);
+ } catch (InterruptedException e) {
+ log.info("Interrupted looking for work at path {}; {}",
path, event, e);
+ }
+ } else {
+ log.info("Unexpected path for NodeChildrenChanged event
watching path {}; {}", path,
+ event);
}
- } else {
- log.info("Unexpected path for NodeChildrenChanged event watching
path {}; {}", path,
- event);
- }
- break;
- default:
- log.info("Unexpected event watching path {}; {}", path, event);
- break;
+ break;
+ default:
+ log.info("Unexpected event watching path {}; {}", path, event);
+ break;
+ }
}
- }
- });
+ });
+ } else {
+ children = zoo.getChildren(path);
+ }
+
+ lookForWork(processor, children, executor, maxThreads);
+
+ }
+
+ /**
+ * Calls {@code runOne} to attempt to process all currently available work,
then adds a background
+ * thread that looks for work in the future.
+ */
+ public void processExistingAndFuture(final Processor processor,
+ ThreadPoolExecutor executorService) throws KeeperException,
InterruptedException {
- lookForWork(processor, children);
+ processExistingWork(processor, executorService,
executorService.getCorePoolSize(), true);
// Add a little jitter to avoid all the tservers slamming zookeeper at once
ThreadPools.watchCriticalScheduledTask(
@@ -228,7 +258,8 @@ public class DistributedWorkQueue {
public void run() {
log.debug("Looking for work in {}", path);
try {
- lookForWork(processor, zoo.getChildren(path));
+ lookForWork(processor, zoo.getChildren(path), executorService,
+ executorService.getCorePoolSize());
} catch (KeeperException e) {
log.error("Failed to look for work", e);
} catch (InterruptedException e) {
diff --git a/server/compactor/pom.xml b/server/compactor/pom.xml
index eaa30280f8..0f021dd497 100644
--- a/server/compactor/pom.xml
+++ b/server/compactor/pom.xml
@@ -55,6 +55,10 @@
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-start</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.accumulo</groupId>
+ <artifactId>accumulo-tserver</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client-api</artifactId>
diff --git
a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
index 2d93e42068..6a234824db 100644
---
a/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
+++
b/server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java
@@ -109,6 +109,7 @@ import org.apache.accumulo.server.rpc.ServerAddress;
import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.rpc.ThriftProcessorTypes;
import org.apache.accumulo.server.security.SecurityOperation;
+import org.apache.accumulo.tserver.log.LogSorter;
import org.apache.hadoop.fs.Path;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;
@@ -635,6 +636,8 @@ public class Compactor extends AbstractServer implements
MetricsProducer, Compac
final AtomicReference<Throwable> err = new AtomicReference<>();
final AtomicLong timeSinceLastCompletion = new AtomicLong(0L);
+ final LogSorter logSorter = new LogSorter(getContext(),
getConfiguration());
+ long nextSortLogsCheckTime = System.currentTimeMillis();
while (!shutdown) {
@@ -649,6 +652,14 @@ public class Compactor extends AbstractServer implements
MetricsProducer, Compac
err.set(null);
JOB_HOLDER.reset();
+ if (System.currentTimeMillis() > nextSortLogsCheckTime) {
+ // Attempt to process all existing log sorting work serially in this
thread.
+ // When no work remains, this call will return so that we can look
for compaction
+ // work.
+ LOG.debug("Checking to see if any recovery logs need sorting");
+ nextSortLogsCheckTime = logSorter.sortLogsIfNeeded();
+ }
+
TExternalCompactionJob job;
try {
job = getNextJob(getNextId());
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java
index 2efd38b709..a5d74b3c07 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java
@@ -103,6 +103,7 @@ import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.rpc.ThriftProcessorTypes;
import org.apache.accumulo.server.security.SecurityUtil;
import
org.apache.accumulo.tserver.TabletServerResourceManager.TabletResourceManager;
+import org.apache.accumulo.tserver.log.LogSorter;
import org.apache.accumulo.tserver.metrics.TabletServerScanMetrics;
import org.apache.accumulo.tserver.session.MultiScanSession;
import org.apache.accumulo.tserver.session.ScanSession;
@@ -390,6 +391,22 @@ public class ScanServer extends AbstractServer
ServiceLock lock = announceExistence();
+ int threadPoolSize =
getConfiguration().getCount(Property.SSERV_WAL_SORT_MAX_CONCURRENT);
+ if (threadPoolSize > 0) {
+ final LogSorter logSorter = new LogSorter(context, getConfiguration());
+ try {
+ // Attempt to process all existing log sorting work and start a
background
+ // thread to look for log sorting work in the future
+ logSorter.startWatchingForRecoveryLogs(threadPoolSize);
+ } catch (Exception ex) {
+ log.error("Error starting LogSorter");
+ throw new RuntimeException(ex);
+ }
+ } else {
+ log.info(
+ "Log sorting for tablet recovery is disabled,
SSERV_WAL_SORT_MAX_CONCURRENT is less than 1.");
+ }
+
try {
while (!serverStopRequested) {
UtilWaitThread.sleep(1000);
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 90244c3ba1..0de617455c 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -602,12 +602,22 @@ public class TabletServer extends AbstractServer
implements TabletHostingServer
throw new RuntimeException(e);
}
- try {
- logSorter.startWatchingForRecoveryLogs();
- } catch (Exception ex) {
- log.error("Error setting watches for recoveries");
- throw new RuntimeException(ex);
+ int threadPoolSize =
+
getContext().getConfiguration().getCount(Property.TSERV_WAL_SORT_MAX_CONCURRENT);
+ if (threadPoolSize > 0) {
+ try {
+ // Attempt to process all existing log sorting work and start a
background
+ // thread to look for log sorting work in the future
+ logSorter.startWatchingForRecoveryLogs(threadPoolSize);
+ } catch (Exception ex) {
+ log.error("Error starting LogSorter");
+ throw new RuntimeException(ex);
+ }
+ } else {
+ log.info(
+ "Log sorting for tablet recovery is disabled,
TSERV_WAL_SORT_MAX_CONCURRENT is less than 1.");
}
+
final AccumuloConfiguration aconf = getConfiguration();
final long onDemandUnloaderInterval =
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
index 33ce1989e4..dda829d0a8 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java
@@ -62,6 +62,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.MoreExecutors;
public class LogSorter {
@@ -290,12 +291,30 @@ public class LogSorter {
}
}
- public void startWatchingForRecoveryLogs() throws KeeperException,
InterruptedException {
- int threadPoolSize =
this.conf.getCount(Property.TSERV_WAL_SORT_MAX_CONCURRENT);
+ /**
+ * Sort any logs that need sorting in the current thread.
+ *
+ * @return The time in millis when the next check can be done.
+ */
+ public long sortLogsIfNeeded() throws KeeperException, InterruptedException {
+ DistributedWorkQueue dwq = new DistributedWorkQueue(
+ context.getZooKeeperRoot() + Constants.ZRECOVERY, sortedLogConf,
context);
+ dwq.processExistingWork(new LogProcessor(),
MoreExecutors.newDirectExecutorService(), 1, false);
+ return System.currentTimeMillis() + dwq.getCheckInterval();
+ }
+
+ /**
+ * Sort any logs that need sorting in a ThreadPool using
+ * {@link Property#TSERV_WAL_SORT_MAX_CONCURRENT} threads. This method will
start a background
+ * thread to look for log sorting work in the future that will be processed
by the
+ * ThreadPoolExecutor
+ */
+ public void startWatchingForRecoveryLogs(int threadPoolSize)
+ throws KeeperException, InterruptedException {
ThreadPoolExecutor threadPool = ThreadPools.getServerThreadPools()
.createFixedThreadPool(threadPoolSize, this.getClass().getName(),
true);
new DistributedWorkQueue(context.getZooKeeperRoot() + Constants.ZRECOVERY,
sortedLogConf,
- context).startProcessing(new LogProcessor(), threadPool);
+ context).processExistingAndFuture(new LogProcessor(), threadPool);
}
public List<RecoveryStatus> getLogSorts() {
diff --git a/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
b/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
new file mode 100644
index 0000000000..c8fb440eea
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/RecoveryIT.java
@@ -0,0 +1,231 @@
+/*
+ * 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.test;
+
+import static org.apache.accumulo.harness.AccumuloITBase.MINI_CLUSTER_ONLY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.stream.IntStream;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+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.data.TableId;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
+import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
+import org.apache.accumulo.core.spi.balancer.TableLoadBalancer;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
+import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterControl;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.log.SortedLogState;
+import org.apache.accumulo.server.manager.recovery.RecoveryPath;
+import org.apache.accumulo.test.functional.ReadWriteIT;
+import org.apache.accumulo.test.util.Wait;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+@Tag(MINI_CLUSTER_ONLY)
+public class RecoveryIT extends AccumuloClusterHarness {
+
+ private static final String RESOURCE_GROUP = "RG1";
+
+ private volatile boolean disableTabletServerLogSorting = false;
+
+ @Override
+ protected Duration defaultTimeout() {
+ return Duration.ofMinutes(3);
+ }
+
+ @Override
+ public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration
hadoopCoreSite) {
+ cfg.getClusterServerConfiguration().setNumDefaultTabletServers(1);
+ cfg.setProperty(Property.MANAGER_TABLET_GROUP_WATCHER_INTERVAL, "5s");
+
cfg.getClusterServerConfiguration().addTabletServerResourceGroup(RESOURCE_GROUP,
3);
+ cfg.setProperty(Property.INSTANCE_ZK_TIMEOUT, "15s");
+ if (disableTabletServerLogSorting) {
+ cfg.setProperty(Property.TSERV_WAL_SORT_MAX_CONCURRENT, "0");
+ }
+ // file system supports recovery
+ hadoopCoreSite.set("fs.file.impl", RawLocalFileSystem.class.getName());
+ }
+
+ @Override
+ public void setupCluster() throws Exception {
+ // Do *NOT* startup the cluster here. We are doing it in the test
+ // method so that we can modify the properties for each test run
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"TSERVER", "SSERVER", "COMPACTOR"})
+ public void test(String serverForSorting) throws Exception {
+
+ // Determine whether or not we need to disable log sorting
+ // in the TabletServer. We want to do this when the serverForSorting
+ // parameter is SSERVER or COMPACTOR
+ switch (serverForSorting) {
+ case "TSERVER":
+ disableTabletServerLogSorting = false;
+ break;
+ case "SSERVER":
+ case "COMPACTOR":
+ default:
+ disableTabletServerLogSorting = true;
+ }
+
+ // Start the cluster
+ super.setupCluster();
+
+ // create a table
+ String tableName = getUniqueNames(1)[0];
+ try (AccumuloClient c =
Accumulo.newClient().from(getClientProps()).build()) {
+
+ SortedSet<Text> splits = new TreeSet<>();
+ IntStream.range(97, 122).forEach(i -> splits.add(new Text(new String(""
+ i))));
+
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ Map<String,String> tableProps = new HashMap<>();
+ tableProps.put(Property.TABLE_MAJC_RATIO.getKey(), "100");
+ tableProps.put(Property.TABLE_FILE_MAX.getKey(), "3");
+ tableProps.put(TableLoadBalancer.TABLE_ASSIGNMENT_GROUP_PROPERTY,
RESOURCE_GROUP);
+ ntc.setProperties(tableProps);
+ ntc.withSplits(splits);
+ ntc.withInitialTabletAvailability(TabletAvailability.ONDEMAND);
+ c.tableOperations().create(tableName, ntc);
+
+ c.instanceOperations().waitForBalance();
+
+ TableId tid =
TableId.of(c.tableOperations().tableIdMap().get(tableName));
+
+ ReadWriteIT.ingest(c, 1000, 1, 1, 0, "", tableName, 100);
+
+ // Confirm that there are no walog entries for this table
+ assertEquals(0, countWaLogEntries(c, tid));
+
+ MiniAccumuloClusterImpl mini = (MiniAccumuloClusterImpl) cluster;
+ MiniAccumuloClusterControl control = (MiniAccumuloClusterControl)
cluster.getClusterControl();
+
+ // Stop any running Compactors and ScanServers
+ control.stopAllServers(ServerType.COMPACTOR);
+ control.stopAllServers(ServerType.SCAN_SERVER);
+
+ // Kill the TabletServer in resource group that is hosting the table
+ List<Process> procs = control.getTabletServers(RESOURCE_GROUP);
+ assertEquals(3, procs.size());
+ for (int i = 0; i < 3; i++) {
+ procs.get(i).destroyForcibly().waitFor();
+ }
+ control.getTabletServers(RESOURCE_GROUP).clear();
+
+ // The TabletGroupWatcher in the Manager will notice that the
TabletServer is dead
+ // and will assign the TabletServer's walog to all of the tablets that
were assigned
+ // to that server. Confirm that walog entries exist for this tablet
+ if (!serverForSorting.equals("TSERVER")) {
+ Wait.waitFor(() -> countWaLogEntries(c, tid) > 0, 60_000);
+ }
+
+ // Start the process that will do the log sorting
+ switch (serverForSorting) {
+ case "TSERVER":
+ // We don't need to re-start the TabletServers here, there is
+ // already a TabletServer running in the default group that
+ // is hosting the root and metadata tables. It should perform
+ // the log sorting.
+ break;
+ case "SSERVER":
+
mini.getConfig().getClusterServerConfiguration().setNumDefaultScanServers(1);
+ getClusterControl().startAllServers(ServerType.SCAN_SERVER);
+ break;
+ case "COMPACTOR":
+
mini.getConfig().getClusterServerConfiguration().setNumDefaultCompactors(1);
+ getClusterControl().startAllServers(ServerType.COMPACTOR);
+ break;
+ case "ALL":
+ default:
+ fail("Unhandled server type: " + serverForSorting);
+ }
+
+ // Confirm sorting completed
+ Wait.waitFor(() -> logSortingCompleted(c, tid) == true, 60_000);
+
+ // Start the tablet servers so that the Manager
+ // can assign the table and so that recovery can be completed.
+ getClusterControl().startAllServers(ServerType.TABLET_SERVER);
+
+ c.instanceOperations().waitForBalance();
+
+ // When the tablet is hosted, the sorted walogs will be applied
+ // Confirm the 3 walog entries are gone for this tablet
+ assertEquals(0, countWaLogEntries(c, tid));
+
+ // Scan the data and make sure its there.
+ ReadWriteIT.verify(c, 1000, 1, 1, 0, "", tableName);
+
+ }
+ }
+
+ private long countWaLogEntries(AccumuloClient c, TableId tableId) throws
Exception {
+ try (TabletsMetadata tabletsMetadata = ((ClientContext)
c).getAmple().readTablets()
+ .forTable(tableId).fetch(TabletMetadata.ColumnType.LOGS).build()) {
+ return tabletsMetadata.stream().filter(tabletMetadata ->
tabletMetadata.getLogs() != null)
+ .count();
+ }
+ }
+
+ private boolean logSortingCompleted(AccumuloClient c, TableId tableId)
throws Exception {
+ try (TabletsMetadata tabletsMetadata = ((ClientContext)
c).getAmple().readTablets()
+ .forTable(tableId).fetch(TabletMetadata.ColumnType.LOGS).build()) {
+ ServerContext ctx = getCluster().getServerContext();
+ for (TabletMetadata tm : tabletsMetadata) {
+ for (LogEntry walog : tm.getLogs()) {
+ String sortId = walog.getUniqueID().toString();
+ String filename = walog.getPath();
+ String dest = RecoveryPath.getRecoveryPath(new
Path(filename)).toString();
+
+ if (ctx.getZooCache().get(ctx.getZooKeeperRoot() +
Constants.ZRECOVERY + "/" + sortId)
+ != null
+ ||
!ctx.getVolumeManager().exists(SortedLogState.getFinishedMarkerPath(dest))) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+ }
+
+}
