You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by "dlmarion (via GitHub)" <gi...@apache.org> on 2024/02/06 21:58:33 UTC
[PR] Add LogSorter to Compactor and ScanServer [accumulo]
dlmarion opened a new pull request, #4239:
URL: https://github.com/apache/accumulo/pull/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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1488545325
##########
server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java:
##########
@@ -651,10 +652,13 @@ public void run() {
err.set(null);
JOB_HOLDER.reset();
- // 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.
- logSorter.sortLogsIfNeeded();
+ if (System.nanoTime() > nextSortLogsCheckTime) {
Review Comment:
Usually need to subtract two nano times to compute a duration when using nano time. Iits not always safe directly compare nano times because it can wrap, going from something like Long.MAX-10 to Long.MAX-10+20 which would go from positive to negative. I am not actually sure if there is a problem here, just know this does not follow the usual pattern of using nano time to compute a duration. When computing durations those should always be positive.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1483679807
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java:
##########
@@ -292,11 +289,27 @@ void writeBuffer(String destPath, List<Pair<LogFileKey,LogFileValue>> buffer, in
}
}
- public void startWatchingForRecoveryLogs(ThreadPoolExecutor distWorkQThreadPool)
+ /**
+ * Sort any logs that need sorting in the current thread.
+ */
+ public void sortLogsIfNeeded() throws KeeperException, InterruptedException {
+ new DistributedWorkQueue(context.getZooKeeperRoot() + Constants.ZRECOVERY, sortedLogConf,
Review Comment:
In the distributed workqueue [constructor](https://github.com/apache/accumulo/blob/113e495dbff884309c4213b9f5016c789850fe42/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java#L172-L173) it sets the period it will look for work to : `60 sec + rand.nextInt(60)`. Could do something similar here where this function would not execute more than once every `60 sec + rand.nextInt(60)`. I assume this was done to avoid lots of request to ZK.
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java:
##########
@@ -390,6 +391,22 @@ public void run() {
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.warn(
Review Comment:
If a user explicitly disabled this not sure it should log a warn.
```suggestion
log.info(
```
##########
test/src/main/java/org/apache/accumulo/test/RecoveryIT.java:
##########
@@ -0,0 +1,226 @@
+/*
+ * 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 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.BatchWriter;
+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.Mutation;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+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.util.Wait;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+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(2);
+ }
+
+ @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, 1);
+ 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()) {
+ 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.withInitialTabletAvailability(TabletAvailability.ONDEMAND);
+ c.tableOperations().create(tableName, ntc);
+
+ TableId tid = TableId.of(c.tableOperations().tableIdMap().get(tableName));
+
+ // create 3 flush files
+ try (BatchWriter bw = c.createBatchWriter(tableName)) {
+ Mutation m = new Mutation("a");
+ m.put("b", "c", new Value("v"));
+ for (int i = 0; i < 3; i++) {
+ bw.addMutation(m);
+ bw.flush();
+ c.tableOperations().flush(tableName, null, null, true);
+ }
+ // create an unsaved mutation
+ bw.addMutation(m);
+ }
+
+ // 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(1, procs.size());
+ procs.get(0).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
+ 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));
+
Review Comment:
Would be good to scan the tablet here to see if the data written by the batch writer is present.
##########
test/src/main/java/org/apache/accumulo/test/RecoveryIT.java:
##########
@@ -0,0 +1,226 @@
+/*
+ * 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 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.BatchWriter;
+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.Mutation;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+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.util.Wait;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+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(2);
+ }
+
+ @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, 1);
+ 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()) {
+ 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.withInitialTabletAvailability(TabletAvailability.ONDEMAND);
+ c.tableOperations().create(tableName, ntc);
+
+ TableId tid = TableId.of(c.tableOperations().tableIdMap().get(tableName));
+
+ // create 3 flush files
+ try (BatchWriter bw = c.createBatchWriter(tableName)) {
+ Mutation m = new Mutation("a");
+ m.put("b", "c", new Value("v"));
+ for (int i = 0; i < 3; i++) {
+ bw.addMutation(m);
+ bw.flush();
+ c.tableOperations().flush(tableName, null, null, true);
+ }
+ // create an unsaved mutation
+ bw.addMutation(m);
+ }
+
+ // 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(1, procs.size());
+ procs.get(0).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
+ Wait.waitFor(() -> countWaLogEntries(c, tid) > 0, 60_000);
Review Comment:
For the TSERVER test case logs could be set on the manager and recovered by the tserver w/o this code seeing it.
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -614,18 +614,28 @@ public void run() {
new DistributedWorkQueue(getContext().getZooKeeperRoot() + Constants.ZBULK_FAILED_COPYQ,
getConfiguration(), getContext());
try {
- bulkFailedCopyQ.startProcessing(new BulkFailedCopyProcessor(getContext()),
+ bulkFailedCopyQ.processExistingAndFuture(new BulkFailedCopyProcessor(getContext()),
distWorkQThreadPool);
} catch (Exception e1) {
throw new RuntimeException("Failed to start distributed work queue for copying ", e1);
}
- try {
- logSorter.startWatchingForRecoveryLogs(distWorkQThreadPool);
- } 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.warn(
Review Comment:
```suggestion
log.info(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1484674245
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java:
##########
@@ -292,11 +289,27 @@ void writeBuffer(String destPath, List<Pair<LogFileKey,LogFileValue>> buffer, in
}
}
- public void startWatchingForRecoveryLogs(ThreadPoolExecutor distWorkQThreadPool)
+ /**
+ * Sort any logs that need sorting in the current thread.
+ */
+ public void sortLogsIfNeeded() throws KeeperException, InterruptedException {
+ new DistributedWorkQueue(context.getZooKeeperRoot() + Constants.ZRECOVERY, sortedLogConf,
Review Comment:
Those values are only used when it creates the background thread in `processExistingAndFuture`. The TabletServer and ScanServer call that method. The Compactor calls `processExistingWork` which does not set up a background thread to poll ZK for work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1484697463
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java:
##########
@@ -292,11 +289,27 @@ void writeBuffer(String destPath, List<Pair<LogFileKey,LogFileValue>> buffer, in
}
}
- public void startWatchingForRecoveryLogs(ThreadPoolExecutor distWorkQThreadPool)
+ /**
+ * Sort any logs that need sorting in the current thread.
+ */
+ public void sortLogsIfNeeded() throws KeeperException, InterruptedException {
+ new DistributedWorkQueue(context.getZooKeeperRoot() + Constants.ZRECOVERY, sortedLogConf,
Review Comment:
I realize the implementations are very different. Was just thinking the different impls could possibly have same minimum poll rate for ZK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1484431806
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java:
##########
@@ -292,11 +289,27 @@ void writeBuffer(String destPath, List<Pair<LogFileKey,LogFileValue>> buffer, in
}
}
- public void startWatchingForRecoveryLogs(ThreadPoolExecutor distWorkQThreadPool)
+ /**
+ * Sort any logs that need sorting in the current thread.
+ */
+ public void sortLogsIfNeeded() throws KeeperException, InterruptedException {
+ new DistributedWorkQueue(context.getZooKeeperRoot() + Constants.ZRECOVERY, sortedLogConf,
Review Comment:
This method was added for the Compactor. Instead of the Compactor calling the other method that starts a background thread to watch for future log sorting work, it calls this method to process any log sorting work before asking for the next compaction job. In this case I don't think we want to wait, we want to execute immediate any log sorting work in the Compactor thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1489436392
##########
server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java:
##########
@@ -651,10 +652,13 @@ public void run() {
err.set(null);
JOB_HOLDER.reset();
- // 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.
- logSorter.sortLogsIfNeeded();
+ if (System.nanoTime() > nextSortLogsCheckTime) {
Review Comment:
I changed this to millis to avoid doing something incorrect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1484564699
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java:
##########
@@ -292,11 +289,27 @@ void writeBuffer(String destPath, List<Pair<LogFileKey,LogFileValue>> buffer, in
}
}
- public void startWatchingForRecoveryLogs(ThreadPoolExecutor distWorkQThreadPool)
+ /**
+ * Sort any logs that need sorting in the current thread.
+ */
+ public void sortLogsIfNeeded() throws KeeperException, InterruptedException {
+ new DistributedWorkQueue(context.getZooKeeperRoot() + Constants.ZRECOVERY, sortedLogConf,
Review Comment:
Was not thinking of waiting, was just thinking of making the rate at which ZK is polled for recovery work consistent across server types. Something like the following.
```suggestion
if((currTime - lastCheckTime) > checkThreshold){
// the checkThreshold exists to so that the compactor polls ZK for recovery work at a rate that is similar to tservers and sservers
new DistributedWorkQueue(context.getZooKeeperRoot() + Constants.ZRECOVERY, sortedLogConf,....);
lastCheckTime = currTime;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1484502659
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java:
##########
@@ -390,6 +391,22 @@ public void run() {
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.warn(
Review Comment:
Resolved in the e937629 merge commit from main to elasticity
##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -614,18 +614,28 @@ public void run() {
new DistributedWorkQueue(getContext().getZooKeeperRoot() + Constants.ZBULK_FAILED_COPYQ,
getConfiguration(), getContext());
try {
- bulkFailedCopyQ.startProcessing(new BulkFailedCopyProcessor(getContext()),
+ bulkFailedCopyQ.processExistingAndFuture(new BulkFailedCopyProcessor(getContext()),
distWorkQThreadPool);
} catch (Exception e1) {
throw new RuntimeException("Failed to start distributed work queue for copying ", e1);
}
- try {
- logSorter.startWatchingForRecoveryLogs(distWorkQThreadPool);
- } 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.warn(
Review Comment:
Resolved in the e937629 merge commit from main to elasticity
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1484569093
##########
test/src/main/java/org/apache/accumulo/test/RecoveryIT.java:
##########
@@ -0,0 +1,226 @@
+/*
+ * 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 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.BatchWriter;
+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.Mutation;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.data.Value;
+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.util.Wait;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+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(2);
+ }
+
+ @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, 1);
+ 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()) {
+ 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.withInitialTabletAvailability(TabletAvailability.ONDEMAND);
+ c.tableOperations().create(tableName, ntc);
+
+ TableId tid = TableId.of(c.tableOperations().tableIdMap().get(tableName));
+
+ // create 3 flush files
+ try (BatchWriter bw = c.createBatchWriter(tableName)) {
+ Mutation m = new Mutation("a");
+ m.put("b", "c", new Value("v"));
+ for (int i = 0; i < 3; i++) {
+ bw.addMutation(m);
+ bw.flush();
+ c.tableOperations().flush(tableName, null, null, true);
+ }
+ // create an unsaved mutation
+ bw.addMutation(m);
+ }
+
+ // 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(1, procs.size());
+ procs.get(0).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
+ Wait.waitFor(() -> countWaLogEntries(c, tid) > 0, 60_000);
Review Comment:
Wonder if the following would work
```suggestion
if(!serverForSortin.equals("TSERVER")){
Wait.waitFor(() -> countWaLogEntries(c, tid) > 0, 60_000);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #4239:
URL: https://github.com/apache/accumulo/pull/4239#discussion_r1484584840
##########
test/src/main/java/org/apache/accumulo/test/functional/ManyWriteAheadLogsIT.java:
##########
@@ -92,8 +93,32 @@ public void alterConfig() throws Exception {
* not single tablet references a lot of write ahead logs. Want to ensure the tablet server forces
* minor compactions for this situation.
*/
- @Test
- public void testMany() throws Exception {
+ @ParameterizedTest
+ @ValueSource(strings = {"TSERVER", "SSERVER", "COMPACTOR", "ALL"})
+ public void testMany(String testType) throws Exception {
+
+ switch (testType) {
+ case "TSERVER":
+ // Stop everything but the tablet server
+ getClusterControl().stopAllServers(ServerType.COMPACTOR);
+ getClusterControl().stopAllServers(ServerType.SCAN_SERVER);
+ break;
+ case "SSERVER":
+ // Stop everything but the scan server
+ getClusterControl().stopAllServers(ServerType.TABLET_SERVER);
Review Comment:
How do the batch writes done later work if all tservers are stopped?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add LogSorter to Compactor and ScanServer [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion merged PR #4239:
URL: https://github.com/apache/accumulo/pull/4239
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org