This is an automated email from the ASF dual-hosted git repository.
kturner 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 12c6433acb tests log recovery using FlakyAmple (#4762)
12c6433acb is described below
commit 12c6433acb4b734e654594e76293710a5e4bd17e
Author: Keith Turner <[email protected]>
AuthorDate: Fri Jul 26 08:33:38 2024 -0700
tests log recovery using FlakyAmple (#4762)
Modfied a log recovery test to run using FlakyAmple
This will provide coverage for the case of flay
conditional mutations that are written to tablets
as part log recovery.
Most of the code in WalSunnyDayIT was moved to
WalSunnyDatBaseIT w/o modification. So even
though this commit has lots of diff, there are
very little actual changes.
This change is for #4710
---
.../accumulo/test/functional/WALFlakyAmpleIT.java | 34 +++
.../{WALSunnyDayIT.java => WALSunnyDayBaseIT.java} | 5 +-
.../accumulo/test/functional/WALSunnyDayIT.java | 269 +--------------------
3 files changed, 42 insertions(+), 266 deletions(-)
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/WALFlakyAmpleIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/WALFlakyAmpleIT.java
new file mode 100644
index 0000000000..ac6d1373e4
--- /dev/null
+++
b/test/src/main/java/org/apache/accumulo/test/functional/WALFlakyAmpleIT.java
@@ -0,0 +1,34 @@
+/*
+ * 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.functional;
+
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.ample.FlakyAmpleManager;
+import org.apache.accumulo.test.ample.FlakyAmpleTserver;
+import org.apache.hadoop.conf.Configuration;
+
+public class WALFlakyAmpleIT extends WALSunnyDayBaseIT {
+ @Override
+ protected void configure(MiniAccumuloConfigImpl cfg, Configuration
hadoopCoreSite) {
+ super.configure(cfg, hadoopCoreSite);
+ cfg.setServerClass(ServerType.TABLET_SERVER, FlakyAmpleTserver.class);
+ cfg.setServerClass(ServerType.MANAGER, FlakyAmpleManager.class);
+ }
+}
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayBaseIT.java
similarity index 98%
copy from
test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayIT.java
copy to
test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayBaseIT.java
index 2d085c8893..b64d539c9a 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayBaseIT.java
@@ -26,7 +26,6 @@ import static
org.apache.accumulo.core.conf.Property.TSERV_WAL_MAX_SIZE;
import static org.apache.accumulo.core.conf.Property.TSERV_WAL_REPLICATION;
import static org.apache.accumulo.core.security.Authorizations.EMPTY;
import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
-import static org.apache.accumulo.harness.AccumuloITBase.SUNNY_DAY;
import static org.apache.accumulo.minicluster.ServerType.GARBAGE_COLLECTOR;
import static org.apache.accumulo.minicluster.ServerType.TABLET_SERVER;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -67,13 +66,11 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.Text;
import org.apache.zookeeper.KeeperException.NoNodeException;
-import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import com.google.common.collect.Iterators;
-@Tag(SUNNY_DAY)
-public class WALSunnyDayIT extends ConfigurableMacBase {
+public abstract class WALSunnyDayBaseIT extends ConfigurableMacBase {
private static final Text CF = new Text(new byte[0]);
diff --git
a/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayIT.java
b/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayIT.java
index 2d085c8893..78bb2cf7e0 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/WALSunnyDayIT.java
@@ -18,272 +18,17 @@
*/
package org.apache.accumulo.test.functional;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.accumulo.core.conf.Property.GC_CYCLE_DELAY;
-import static org.apache.accumulo.core.conf.Property.GC_CYCLE_START;
-import static org.apache.accumulo.core.conf.Property.INSTANCE_ZK_TIMEOUT;
-import static org.apache.accumulo.core.conf.Property.TSERV_WAL_MAX_SIZE;
-import static org.apache.accumulo.core.conf.Property.TSERV_WAL_REPLICATION;
-import static org.apache.accumulo.core.security.Authorizations.EMPTY;
-import static org.apache.accumulo.core.util.LazySingletons.RANDOM;
import static org.apache.accumulo.harness.AccumuloITBase.SUNNY_DAY;
-import static org.apache.accumulo.minicluster.ServerType.GARBAGE_COLLECTOR;
-import static org.apache.accumulo.minicluster.ServerType.TABLET_SERVER;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-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.Scanner;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.metadata.AccumuloTable;
-import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
-import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
-import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
-import org.apache.accumulo.manager.state.SetGoalState;
-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.WalStateManager;
-import org.apache.accumulo.server.log.WalStateManager.WalMarkerException;
-import org.apache.accumulo.server.log.WalStateManager.WalState;
-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.apache.zookeeper.KeeperException.NoNodeException;
import org.junit.jupiter.api.Tag;
-import org.junit.jupiter.api.Test;
-
-import com.google.common.collect.Iterators;
+/*
+ * This class is empty because of the SUNNY_DAY tag. WALSunnyDayBaseIT exists
to share code with
+ * WALFlakyAmpleIT. Ideally the code in WALSunnyDayBaseIT could be moved into
this class and
+ * WALFlakyAmpleIT could extend this class. However, if it did then
WALFlakyAmpleIT would inherit
+ * the SUNNY_DAY tag which is not desirable.
+ */
@Tag(SUNNY_DAY)
-public class WALSunnyDayIT extends ConfigurableMacBase {
-
- private static final Text CF = new Text(new byte[0]);
-
- @Override
- protected void configure(MiniAccumuloConfigImpl cfg, Configuration
hadoopCoreSite) {
- cfg.setProperty(GC_CYCLE_DELAY, "1s");
- cfg.setProperty(GC_CYCLE_START, "0s");
- cfg.setProperty(TSERV_WAL_MAX_SIZE, "1M");
- cfg.setProperty(TSERV_WAL_REPLICATION, "1");
- cfg.setProperty(INSTANCE_ZK_TIMEOUT, "15s");
- cfg.getClusterServerConfiguration().setNumDefaultTabletServers(1);
- hadoopCoreSite.set("fs.file.impl", RawLocalFileSystem.class.getName());
- }
-
- int countInUse(Collection<WalState> bools) {
- int result = 0;
- for (WalState b : bools) {
- if (b != WalState.UNREFERENCED) {
- result++;
- }
- }
- return result;
- }
-
- @Test
- public void test() throws Exception {
- MiniAccumuloClusterImpl mac = getCluster();
- MiniAccumuloClusterControl control = mac.getClusterControl();
- control.stop(GARBAGE_COLLECTOR);
- ServerContext context = getServerContext();
- try (AccumuloClient c =
Accumulo.newClient().from(getClientProperties()).build()) {
- String tableName = getUniqueNames(1)[0];
- c.tableOperations().create(tableName);
- writeSomeData(c, tableName, 1, 1);
-
- // wal markers are added lazily
- Map<String,WalState> wals = getWALsAndAssertCount(context, 2);
- assertEquals(2, countInUse(wals.values()), "all WALs should be in use");
-
- // roll log, get a new next
- writeSomeData(c, tableName, 1001, 50);
- Map<String,WalState> walsAfterRoll = getWALsAndAssertCount(context, 3);
- assertTrue(walsAfterRoll.keySet().containsAll(wals.keySet()),
- "new WALs should be a superset of the old WALs");
- assertEquals(3, countInUse(walsAfterRoll.values()), "all WALs should be
in use");
-
- // flush the tables
- for (String table : new String[] {tableName,
AccumuloTable.METADATA.tableName(),
- AccumuloTable.ROOT.tableName(), AccumuloTable.FATE.tableName()}) {
- c.tableOperations().flush(table, null, null, true);
- }
- Thread.sleep(SECONDS.toMillis(1));
- // rolled WAL is no longer in use, but needs to be GC'd
- Map<String,WalState> walsAfterflush = getWALsAndAssertCount(context, 3);
- assertEquals(2, countInUse(walsAfterflush.values()), "inUse should be
2");
-
- // let the GC run for a little bit
- control.start(GARBAGE_COLLECTOR);
- Thread.sleep(SECONDS.toMillis(5));
- // make sure the unused WAL goes away
- getWALsAndAssertCount(context, 2);
- control.stop(GARBAGE_COLLECTOR);
- // restart the tserver, but don't run recovery on all tablets
- control.stop(TABLET_SERVER);
- // this delays recovery on the normal tables
- assertEquals(0, cluster.exec(SetGoalState.class,
"SAFE_MODE").getProcess().waitFor());
- control.start(TABLET_SERVER);
-
- // wait for the metadata table to go back online
- getRecoveryMarkers(c);
- // allow a little time for the manager to notice ASSIGNED_TO_DEAD_SERVER
tablets
- Thread.sleep(SECONDS.toMillis(5));
- Map<KeyExtent,List<String>> markers = getRecoveryMarkers(c);
- // log.debug("markers " + markers);
- // There should be markers for the created table and also the Fate and
ScanRef tables
- assertEquals(3, markers.size(), "three tablets should have markers");
- assertTrue(
- markers.keySet().stream().anyMatch(extent ->
extent.tableId().canonical().equals("1")),
- "tableId of the keyExtent should be 1");
- assertTrue(
- markers.keySet().stream()
- .anyMatch(extent ->
extent.tableId().equals(AccumuloTable.FATE.tableId())),
- "tableId of the Fate table can't be found");
- assertTrue(
- markers.keySet().stream()
- .anyMatch(extent ->
extent.tableId().equals(AccumuloTable.SCAN_REF.tableId())),
- "tableId of the ScanRef table can't be found");
-
- // put some data in the WAL
- assertEquals(0, cluster.exec(SetGoalState.class,
"NORMAL").getProcess().waitFor());
- verifySomeData(c, tableName, 1001 * 50 + 1);
- writeSomeData(c, tableName, 100, 100);
-
- Map<String,WalState> walsAfterRestart = getWALsAndAssertCount(context,
4);
- // log.debug("wals after " + walsAfterRestart);
- assertEquals(4, countInUse(walsAfterRestart.values()), "used WALs after
restart should be 4");
- control.start(GARBAGE_COLLECTOR);
- Thread.sleep(SECONDS.toMillis(5));
- Map<String,WalState> walsAfterRestartAndGC =
getWALsAndAssertCount(context, 2);
- assertEquals(2, countInUse(walsAfterRestartAndGC.values()), "logs in use
should be 2");
- }
- }
-
- private void verifySomeData(AccumuloClient c, String tableName, int
expected) throws Exception {
- try (Scanner scan = c.createScanner(tableName, EMPTY)) {
- int result = Iterators.size(scan.iterator());
- assertEquals(expected, result);
- }
- }
-
- private void writeSomeData(AccumuloClient client, String tableName, int row,
int col)
- throws Exception {
- try (BatchWriter bw = client.createBatchWriter(tableName)) {
- byte[] rowData = new byte[10];
- byte[] cq = new byte[10];
- byte[] value = new byte[10];
-
- for (int r = 0; r < row; r++) {
- RANDOM.get().nextBytes(rowData);
- Mutation m = new Mutation(rowData);
- for (int c = 0; c < col; c++) {
- RANDOM.get().nextBytes(cq);
- RANDOM.get().nextBytes(value);
- m.put(CF, new Text(cq), new Value(value));
- }
- bw.addMutation(m);
- if (r % 100 == 0) {
- bw.flush();
- }
- }
- }
- }
-
- private Map<KeyExtent,List<String>> getRecoveryMarkers(AccumuloClient c)
throws Exception {
- Map<KeyExtent,List<String>> result = new HashMap<>();
- try (Scanner root = c.createScanner(AccumuloTable.ROOT.tableName(), EMPTY);
- Scanner meta = c.createScanner(AccumuloTable.METADATA.tableName(),
EMPTY)) {
- root.setRange(TabletsSection.getRange());
- root.fetchColumnFamily(LogColumnFamily.NAME);
- TabletColumnFamily.PREV_ROW_COLUMN.fetch(root);
-
- meta.setRange(TabletsSection.getRange());
- meta.fetchColumnFamily(LogColumnFamily.NAME);
- TabletColumnFamily.PREV_ROW_COLUMN.fetch(meta);
-
- List<String> logs = new ArrayList<>();
- Iterator<Entry<Key,Value>> both = Iterators.concat(root.iterator(),
meta.iterator());
- while (both.hasNext()) {
- Entry<Key,Value> entry = both.next();
- Key key = entry.getKey();
- if (key.getColumnFamily().equals(LogColumnFamily.NAME)) {
- logs.add(key.getColumnQualifier().toString());
- }
- if (TabletColumnFamily.PREV_ROW_COLUMN.hasColumns(key) &&
!logs.isEmpty()) {
- KeyExtent extent = KeyExtent.fromMetaPrevRow(entry);
- result.put(extent, logs);
- logs = new ArrayList<>();
- }
- }
- }
- return result;
- }
-
- private final int TIMES_TO_COUNT = 20;
- private final int PAUSE_BETWEEN_COUNTS = 100;
-
- private Map<String,WalState> getWALsAndAssertCount(ServerContext c, int
expectedCount)
- throws Exception {
- // see https://issues.apache.org/jira/browse/ACCUMULO-4110. Sometimes this
test counts the logs
- // before
- // the new standby log is actually ready. So let's try a few times before
failing, returning the
- // last
- // wals variable with the the correct count.
- Map<String,WalState> wals = _getWals(c);
- if (wals.size() == expectedCount) {
- return wals;
- }
-
- int waitLonger = Wait.getTimeoutFactor(e -> 1); // default to 1
- for (int i = 1; i <= TIMES_TO_COUNT; i++) {
- Thread.sleep(i * PAUSE_BETWEEN_COUNTS * waitLonger);
- wals = _getWals(c);
- if (wals.size() == expectedCount) {
- return wals;
- }
- }
-
- fail(
- "Unable to get the correct number of WALs, expected " + expectedCount
+ " but got " + wals);
- return new HashMap<>();
- }
-
- static Map<String,WalState> _getWals(ServerContext c) throws Exception {
- while (true) {
- try {
- Map<String,WalState> result = new HashMap<>();
- WalStateManager wals = new WalStateManager(c);
- for (Entry<Path,WalState> entry : wals.getAllState().entrySet()) {
- // WALs are in use if they are not unreferenced
- result.put(entry.getKey().toString(), entry.getValue());
- }
- return result;
- } catch (WalMarkerException wme) {
- if (wme.getCause() instanceof NoNodeException) {
- log.debug("WALs changed while reading, retrying", wme);
- } else {
- throw wme;
- }
- }
- }
- }
+public class WALSunnyDayIT extends WALSunnyDayBaseIT {
}
