You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2022/02/18 11:58:57 UTC
[accumulo] branch main updated: More improvements to ExternalCompactionITs (#2507)
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new 9650e3b More improvements to ExternalCompactionITs (#2507)
9650e3b is described below
commit 9650e3b484bfacc98590ee5fb228b57ed02484b4
Author: Dave Marion <dl...@apache.org>
AuthorDate: Fri Feb 18 06:58:50 2022 -0500
More improvements to ExternalCompactionITs (#2507)
Modified the methods in ExternalCompactionTestUtils that wait for
some condition to only sleep if the condition is not true. Modified
the test methods to use their own compaction queue for each test.
---
.../compaction/ExternalCompactionTestUtils.java | 50 +++++++++++++++++++---
.../test/compaction/ExternalCompaction_1_IT.java | 46 +++++++++++---------
.../test/compaction/ExternalCompaction_2_IT.java | 32 ++++++++------
.../test/compaction/ExternalCompaction_3_IT.java | 7 +--
4 files changed, 94 insertions(+), 41 deletions(-)
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java
index 5d578b3..5225733 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java
@@ -82,6 +82,12 @@ public class ExternalCompactionTestUtils {
public static final int MAX_DATA = 1000;
public static final String QUEUE1 = "DCQ1";
public static final String QUEUE2 = "DCQ2";
+ public static final String QUEUE3 = "DCQ3";
+ public static final String QUEUE4 = "DCQ4";
+ public static final String QUEUE5 = "DCQ5";
+ public static final String QUEUE6 = "DCQ6";
+ public static final String QUEUE7 = "DCQ7";
+ public static final String QUEUE8 = "DCQ8";
public static String row(int r) {
return String.format("r:%04d", r);
@@ -193,11 +199,35 @@ public class ExternalCompactionTestUtils {
cfg.setProperty("tserver.compaction.major.service.cs1.planner",
DefaultCompactionPlanner.class.getName());
cfg.setProperty("tserver.compaction.major.service.cs1.planner.opts.executors",
- "[{'name':'all', 'type': 'external', 'queue': 'DCQ1'}]");
+ "[{'name':'all', 'type': 'external', 'queue': '" + QUEUE1 + "'}]");
cfg.setProperty("tserver.compaction.major.service.cs2.planner",
DefaultCompactionPlanner.class.getName());
cfg.setProperty("tserver.compaction.major.service.cs2.planner.opts.executors",
- "[{'name':'all', 'type': 'external','queue': 'DCQ2'}]");
+ "[{'name':'all', 'type': 'external','queue': '" + QUEUE2 + "'}]");
+ cfg.setProperty("tserver.compaction.major.service.cs3.planner",
+ DefaultCompactionPlanner.class.getName());
+ cfg.setProperty("tserver.compaction.major.service.cs3.planner.opts.executors",
+ "[{'name':'all', 'type': 'external','queue': '" + QUEUE3 + "'}]");
+ cfg.setProperty("tserver.compaction.major.service.cs4.planner",
+ DefaultCompactionPlanner.class.getName());
+ cfg.setProperty("tserver.compaction.major.service.cs4.planner.opts.executors",
+ "[{'name':'all', 'type': 'external','queue': '" + QUEUE4 + "'}]");
+ cfg.setProperty("tserver.compaction.major.service.cs5.planner",
+ DefaultCompactionPlanner.class.getName());
+ cfg.setProperty("tserver.compaction.major.service.cs5.planner.opts.executors",
+ "[{'name':'all', 'type': 'external','queue': '" + QUEUE5 + "'}]");
+ cfg.setProperty("tserver.compaction.major.service.cs6.planner",
+ DefaultCompactionPlanner.class.getName());
+ cfg.setProperty("tserver.compaction.major.service.cs6.planner.opts.executors",
+ "[{'name':'all', 'type': 'external','queue': '" + QUEUE6 + "'}]");
+ cfg.setProperty("tserver.compaction.major.service.cs7.planner",
+ DefaultCompactionPlanner.class.getName());
+ cfg.setProperty("tserver.compaction.major.service.cs7.planner.opts.executors",
+ "[{'name':'all', 'type': 'external','queue': '" + QUEUE7 + "'}]");
+ cfg.setProperty("tserver.compaction.major.service.cs8.planner",
+ DefaultCompactionPlanner.class.getName());
+ cfg.setProperty("tserver.compaction.major.service.cs8.planner.opts.executors",
+ "[{'name':'all', 'type': 'external','queue': '" + QUEUE8 + "'}]");
cfg.setProperty(Property.COMPACTION_COORDINATOR_FINALIZER_COMPLETION_CHECK_INTERVAL, "5s");
cfg.setProperty(Property.COMPACTION_COORDINATOR_DEAD_COMPACTOR_CHECK_INTERVAL, "5s");
cfg.setProperty(Property.COMPACTION_COORDINATOR_TSERVER_COMPACTION_CHECK_INTERVAL, "3s");
@@ -256,11 +286,13 @@ public class ExternalCompactionTestUtils {
TableId tid) {
Set<ExternalCompactionId> ecids = new HashSet<>();
do {
- UtilWaitThread.sleep(50);
try (TabletsMetadata tm =
ctx.getAmple().readTablets().forTable(tid).fetch(ColumnType.ECOMP).build()) {
tm.stream().flatMap(t -> t.getExternalCompactions().keySet().stream()).forEach(ecids::add);
}
+ if (ecids.isEmpty()) {
+ UtilWaitThread.sleep(50);
+ }
} while (ecids.isEmpty());
return ecids;
}
@@ -279,7 +311,9 @@ public class ExternalCompactionTestUtils {
}
}
}
- UtilWaitThread.sleep(250);
+ if (matches == 0) {
+ UtilWaitThread.sleep(50);
+ }
}
return matches;
}
@@ -289,14 +323,18 @@ public class ExternalCompactionTestUtils {
// The running compaction should be removed
TExternalCompactionList running = ExternalCompactionTestUtils.getRunningCompactions(ctx);
while (running.getCompactions() != null) {
- UtilWaitThread.sleep(250);
running = ExternalCompactionTestUtils.getRunningCompactions(ctx);
+ if (running.getCompactions() == null) {
+ UtilWaitThread.sleep(250);
+ }
}
// The compaction should be in the completed list with the expected state
TExternalCompactionList completed = ExternalCompactionTestUtils.getCompletedCompactions(ctx);
while (completed.getCompactions() == null) {
- UtilWaitThread.sleep(250);
completed = ExternalCompactionTestUtils.getCompletedCompactions(ctx);
+ if (completed.getCompactions() == null) {
+ UtilWaitThread.sleep(50);
+ }
}
for (ExternalCompactionId e : ecids) {
TExternalCompaction tec = completed.getCompactions().get(e.canonical());
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java
index 0c62a3f..819b70a 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java
@@ -22,6 +22,12 @@ import static org.apache.accumulo.minicluster.ServerType.TABLET_SERVER;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.MAX_DATA;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE1;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE2;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE3;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE4;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE5;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE6;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE7;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE8;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.compact;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.createTable;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.getFinalStatesForTable;
@@ -208,13 +214,13 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
// Start our TServer that will not commit the compaction
ProcessInfo tserverProcess = getCluster().exec(ExternalCompactionTServer.class);
- createTable(client, table1, "cs1", 2);
+ createTable(client, table1, "cs3", 2);
writeData(client, table1);
- getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE3);
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- compact(client, table1, 2, QUEUE1, false);
+ compact(client, table1, 2, QUEUE3, false);
TableId tid = Tables.getTableId(getCluster().getServerContext(), table1);
// Wait for the compaction to start by waiting for 1 external compaction column
@@ -232,12 +238,12 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
UtilWaitThread.sleep(250);
}
- getCluster().stopProcessWithTimeout(tserverProcess.getProcess(), 30, TimeUnit.SECONDS);
- getCluster().getClusterControl().stop(ServerType.TABLET_SERVER);
// We need to cancel the compaction or delete the table here because we initiate a user
// compaction above in the test. Even though the external compaction was cancelled
// because we split the table, FaTE will continue to queue up a compaction
client.tableOperations().cancelCompaction(table1);
+ getCluster().stopProcessWithTimeout(tserverProcess.getProcess(), 30, TimeUnit.SECONDS);
+ getCluster().getClusterControl().stop(ServerType.TABLET_SERVER);
} finally {
// We stopped the TServer and started our own, restart the original TabletServers
getCluster().getClusterControl().start(ServerType.TABLET_SERVER);
@@ -251,14 +257,14 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
try (AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table1, "cs1", 200);
+ createTable(client, table1, "cs4", 200);
writeData(client, table1);
- getCluster().getClusterControl().startCompactors(Compactor.class, 2, QUEUE1);
+ getCluster().getClusterControl().startCompactors(Compactor.class, 2, QUEUE4);
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- compact(client, table1, 3, QUEUE1, true);
+ compact(client, table1, 3, QUEUE4, true);
verify(client, table1, 3);
}
@@ -268,7 +274,7 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
public void testConfigurer() throws Exception {
String tableName = this.getUniqueNames(1)[0];
- getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE5);
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
try (AccumuloClient client =
@@ -276,7 +282,7 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
Map<String,String> props = Map.of("table.compaction.dispatcher",
SimpleCompactionDispatcher.class.getName(), "table.compaction.dispatcher.opts.service",
- "cs1", Property.TABLE_FILE_COMPRESSION_TYPE.getKey(), "none");
+ "cs5", Property.TABLE_FILE_COMPRESSION_TYPE.getKey(), "none");
NewTableConfiguration ntc = new NewTableConfiguration().setProperties(props);
client.tableOperations().create(tableName, ntc);
@@ -342,11 +348,11 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
String table1 = this.getUniqueNames(1)[0];
try (AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table1, "cs1");
+ createTable(client, table1, "cs6");
writeData(client, table1);
- getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE6);
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- compact(client, table1, 2, QUEUE1, true);
+ compact(client, table1, 2, QUEUE6, true);
verify(client, table1, 2);
IteratorSetting setting = new IteratorSetting(50, "delete", ExtDevNull.class);
@@ -381,11 +387,11 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
try (final AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table3, "cs1");
+ createTable(client, table3, "cs7");
writeData(client, table3);
- getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE7);
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- compact(client, table3, 2, QUEUE1, false);
+ compact(client, table3, 2, QUEUE7, false);
// ExternalCompactionTServer will not commit the compaction. Wait for the
// metadata table entries to show up.
@@ -464,14 +470,14 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
try (final AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE8);
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- createTable(client, tableName, "cs1");
+ createTable(client, tableName, "cs8");
writeData(client, tableName);
// This should create an A file
- compact(client, tableName, 17, QUEUE1, true);
+ compact(client, tableName, 17, QUEUE8, true);
verify(client, tableName, 17);
try (BatchWriter bw = client.createBatchWriter(tableName)) {
@@ -488,7 +494,7 @@ public class ExternalCompaction_1_IT extends SharedMiniClusterBase {
// run a compaction that only compacts F files
IteratorSetting iterSetting = new IteratorSetting(100, TestFilter.class);
// make sure iterator options make it to compactor process
- iterSetting.addOption("expectedQ", QUEUE1);
+ iterSetting.addOption("expectedQ", QUEUE8);
// compact F file w/ different modulus and user pmodulus option for partial compaction
iterSetting.addOption("pmodulus", 19 + "");
CompactionConfig config = new CompactionConfig().setIterators(List.of(iterSetting))
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_2_IT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_2_IT.java
index 2f43b97..b569f89 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_2_IT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_2_IT.java
@@ -20,6 +20,10 @@ package org.apache.accumulo.test.compaction;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.MAX_DATA;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE1;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE2;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE3;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE4;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE5;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.compact;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.confirmCompactionCompleted;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.confirmCompactionRunning;
@@ -148,7 +152,7 @@ public class ExternalCompaction_2_IT extends SharedMiniClusterBase {
try (AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table1, "cs1");
+ createTable(client, table1, "cs2");
// set compaction ratio to 1 so that majc occurs naturally, not user compaction
// user compaction blocks merge
client.tableOperations().setProperty(table1, Property.TABLE_MAJC_RATIO.toString(), "1.0");
@@ -172,8 +176,10 @@ public class ExternalCompaction_2_IT extends SharedMiniClusterBase {
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
TExternalCompactionList metrics2 = getRunningCompactions(getCluster().getServerContext());
while (metrics2.getCompactions() == null) {
- UtilWaitThread.sleep(50);
metrics2 = getRunningCompactions(getCluster().getServerContext());
+ if (metrics2.getCompactions() == null) {
+ UtilWaitThread.sleep(50);
+ }
}
LOG.info("Taking table offline");
client2.tableOperations().offline(table1, false);
@@ -185,7 +191,7 @@ public class ExternalCompaction_2_IT extends SharedMiniClusterBase {
t.start();
// Start the compactor
- getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(Compactor.class, 1, QUEUE2);
// Wait for the compaction to start by waiting for 1 external compaction column
Set<ExternalCompactionId> ecids = ExternalCompactionTestUtils
@@ -213,8 +219,10 @@ public class ExternalCompaction_2_IT extends SharedMiniClusterBase {
// wait for compaction to be committed by tserver or test timeout
long finalStateCount = getFinalStatesForTable(getCluster(), tid).count();
while (finalStateCount > 0) {
- UtilWaitThread.sleep(250);
finalStateCount = getFinalStatesForTable(getCluster(), tid).count();
+ if (finalStateCount > 0) {
+ UtilWaitThread.sleep(50);
+ }
}
// We need to cancel the compaction or delete the table here because we initiate a user
@@ -231,16 +239,16 @@ public class ExternalCompaction_2_IT extends SharedMiniClusterBase {
public void testUserCompactionCancellation() throws Exception {
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE3);
String table1 = this.getUniqueNames(1)[0];
try (AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table1, "cs1");
+ createTable(client, table1, "cs3");
TableId tid = Tables.getTableId(getCluster().getServerContext(), table1);
writeData(client, table1);
- compact(client, table1, 2, QUEUE1, false);
+ compact(client, table1, 2, QUEUE3, false);
// Wait for the compaction to start by waiting for 1 external compaction column
Set<ExternalCompactionId> ecids = ExternalCompactionTestUtils
@@ -267,16 +275,16 @@ public class ExternalCompaction_2_IT extends SharedMiniClusterBase {
public void testDeleteTableCancelsUserExternalCompaction() throws Exception {
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE4);
String table1 = this.getUniqueNames(1)[0];
try (AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table1, "cs1");
+ createTable(client, table1, "cs4");
TableId tid = Tables.getTableId(getCluster().getServerContext(), table1);
writeData(client, table1);
- compact(client, table1, 2, QUEUE1, false);
+ compact(client, table1, 2, QUEUE4, false);
// Wait for the compaction to start by waiting for 1 external compaction column
Set<ExternalCompactionId> ecids =
@@ -297,13 +305,13 @@ public class ExternalCompaction_2_IT extends SharedMiniClusterBase {
@Test
public void testDeleteTableCancelsExternalCompaction() throws Exception {
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE5);
String table1 = this.getUniqueNames(1)[0];
try (AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table1, "cs1");
+ createTable(client, table1, "cs5");
// set compaction ratio to 1 so that majc occurs naturally, not user compaction
// user compaction blocks delete
client.tableOperations().setProperty(table1, Property.TABLE_MAJC_RATIO.toString(), "1.0");
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_3_IT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_3_IT.java
index a0f20aa..30551a0 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_3_IT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_3_IT.java
@@ -19,6 +19,7 @@
package org.apache.accumulo.test.compaction;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE1;
+import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.QUEUE2;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.compact;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.confirmCompactionCompleted;
import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.createTable;
@@ -141,15 +142,15 @@ public class ExternalCompaction_3_IT extends SharedMiniClusterBase {
@Test
public void testCoordinatorRestartsDuringCompaction() throws Exception {
getCluster().getClusterControl().startCoordinator(CompactionCoordinator.class);
- getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE1);
+ getCluster().getClusterControl().startCompactors(ExternalDoNothingCompactor.class, 1, QUEUE2);
String table1 = this.getUniqueNames(1)[0];
try (AccumuloClient client =
Accumulo.newClient().from(getCluster().getClientProperties()).build()) {
- createTable(client, table1, "cs1", 2);
+ createTable(client, table1, "cs2", 2);
writeData(client, table1);
- compact(client, table1, 2, QUEUE1, false);
+ compact(client, table1, 2, QUEUE2, false);
TableId tid = Tables.getTableId(getCluster().getServerContext(), table1);