You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by "keith-turner (via GitHub)" <gi...@apache.org> on 2023/05/22 22:00:23 UTC
[GitHub] [accumulo] keith-turner commented on a diff in pull request #3415: Add more split integration tests
keith-turner commented on code in PR #3415:
URL: https://github.com/apache/accumulo/pull/3415#discussion_r1201205542
##########
test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java:
##########
@@ -205,4 +219,79 @@ public void deleteSplit() throws Exception {
}
}
+ @Test
+ public void concurrentSplit() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+
+ final String tableName = getUniqueNames(1)[0];
+
+ log.debug("Creating table {}", tableName);
+ c.tableOperations().create(tableName);
+
+ final int numRows = 100_000;
+ log.debug("Ingesting {} rows into {}", numRows, tableName);
+ VerifyParams params = new VerifyParams(getClientProps(), tableName, numRows);
+ TestIngest.ingest(c, params);
+
+ log.debug("Verifying {} rows ingested into {}", numRows, tableName);
+ VerifyIngest.verifyIngest(c, params);
+
+ log.debug("Creating futures that add random splits to the table");
+ ExecutorService es = Executors.newFixedThreadPool(10);
+ final int totalFutures = 100;
+ final int splitsPerFuture = 4;
+ final Set<Text> totalSplits = new HashSet<>();
+ List<Callable<Void>> tasks = new ArrayList<>(totalFutures);
+ for (int i = 0; i < totalFutures; i++) {
+ final Pair<Integer,Integer> splitBounds = getRandomSplitBounds(numRows);
+ final TreeSet<Text> splits = TestIngest.getSplitPoints(splitBounds.getFirst().longValue(),
Review Comment:
Looked at the [TestIngest.getSplitPoints method](https://github.com/apache/accumulo/blob/c5f4e2e25406e597f54638ddc596789c96a7501e/test/src/main/java/org/apache/accumulo/test/TestIngest.java#L193-L205) and it seems it has a bug where when `end-start < numsplits` it will set set `splitSize` to zero and infinite loop will happen. In the previous sentence `end`, `start`, `numSplits`, and `splitSize` are variables in `TestIngest.getSplitPoints`.
Locally I tested this and it never printed anything and just chewed up 100% cpu.
```
public static void main(String[] args) {
var splits = TestIngest.getSplitPoints(500, 502, 4);
System.out.println(splits);
}
```
I think the way the random numbers are generated we could get something where `end-start` is less than 4 and then the test would hang.
##########
test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java:
##########
@@ -205,4 +219,79 @@ public void deleteSplit() throws Exception {
}
}
+ @Test
+ public void concurrentSplit() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+
+ final String tableName = getUniqueNames(1)[0];
+
+ log.debug("Creating table {}", tableName);
+ c.tableOperations().create(tableName);
+
+ final int numRows = 100_000;
+ log.debug("Ingesting {} rows into {}", numRows, tableName);
+ VerifyParams params = new VerifyParams(getClientProps(), tableName, numRows);
+ TestIngest.ingest(c, params);
+
+ log.debug("Verifying {} rows ingested into {}", numRows, tableName);
+ VerifyIngest.verifyIngest(c, params);
+
+ log.debug("Creating futures that add random splits to the table");
+ ExecutorService es = Executors.newFixedThreadPool(10);
+ final int totalFutures = 100;
+ final int splitsPerFuture = 4;
+ final Set<Text> totalSplits = new HashSet<>();
+ List<Callable<Void>> tasks = new ArrayList<>(totalFutures);
+ for (int i = 0; i < totalFutures; i++) {
+ final Pair<Integer,Integer> splitBounds = getRandomSplitBounds(numRows);
+ final TreeSet<Text> splits = TestIngest.getSplitPoints(splitBounds.getFirst().longValue(),
+ splitBounds.getSecond().longValue(), splitsPerFuture);
+ totalSplits.addAll(splits);
+ tasks.add(() -> {
+ c.tableOperations().addSplits(tableName, splits);
+ return null;
+ });
+ }
+
+ log.debug("Submitting futures");
+ List<Future<Void>> futures =
+ tasks.parallelStream().map(es::submit).collect(Collectors.toList());
+
+ log.debug("Waiting for futures to complete");
+ for (Future<?> f : futures) {
+ f.get();
+ }
+ es.shutdown();
+
+ final int expectedSplitCount = totalSplits.size();
+ boolean allSplitsHaveBeenSeen = Wait.waitFor(() -> {
+ final int actualSplitCount = c.tableOperations().listSplits(tableName).size();
+ log.debug("Waiting for {} splits to happen. Total splits: {}", expectedSplitCount,
+ actualSplitCount);
+ return actualSplitCount == expectedSplitCount;
+ });
+
+ assertTrue(allSplitsHaveBeenSeen, "Did not see expected number of splits");
+ }
+ }
+
+ /**
+ * Generates a pair of integers that represent the start and end of a range of splits. The start
+ * and end are randomly generated between 0 and upperBound. The start is guaranteed to be less
+ * than the end.
+ *
+ * @param upperBound the upper bound of the range of splits
+ * @return a pair of integers that represent the start and end of a range of splits
+ */
+ private Pair<Integer,Integer> getRandomSplitBounds(int upperBound) {
+ int start = random.nextInt(upperBound);
+ int end = random.nextInt(upperBound);
+ if (start > end) {
Review Comment:
Its possible to generate two equal numbers.
##########
test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java:
##########
@@ -205,4 +219,79 @@ public void deleteSplit() throws Exception {
}
}
+ @Test
+ public void concurrentSplit() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+
+ final String tableName = getUniqueNames(1)[0];
+
+ log.debug("Creating table {}", tableName);
+ c.tableOperations().create(tableName);
+
+ final int numRows = 100_000;
+ log.debug("Ingesting {} rows into {}", numRows, tableName);
+ VerifyParams params = new VerifyParams(getClientProps(), tableName, numRows);
+ TestIngest.ingest(c, params);
+
+ log.debug("Verifying {} rows ingested into {}", numRows, tableName);
+ VerifyIngest.verifyIngest(c, params);
+
+ log.debug("Creating futures that add random splits to the table");
+ ExecutorService es = Executors.newFixedThreadPool(10);
+ final int totalFutures = 100;
+ final int splitsPerFuture = 4;
+ final Set<Text> totalSplits = new HashSet<>();
+ List<Callable<Void>> tasks = new ArrayList<>(totalFutures);
+ for (int i = 0; i < totalFutures; i++) {
+ final Pair<Integer,Integer> splitBounds = getRandomSplitBounds(numRows);
+ final TreeSet<Text> splits = TestIngest.getSplitPoints(splitBounds.getFirst().longValue(),
+ splitBounds.getSecond().longValue(), splitsPerFuture);
+ totalSplits.addAll(splits);
+ tasks.add(() -> {
+ c.tableOperations().addSplits(tableName, splits);
+ return null;
+ });
+ }
+
+ log.debug("Submitting futures");
+ List<Future<Void>> futures =
+ tasks.parallelStream().map(es::submit).collect(Collectors.toList());
+
+ log.debug("Waiting for futures to complete");
+ for (Future<?> f : futures) {
+ f.get();
+ }
+ es.shutdown();
+
+ final int expectedSplitCount = totalSplits.size();
+ boolean allSplitsHaveBeenSeen = Wait.waitFor(() -> {
+ final int actualSplitCount = c.tableOperations().listSplits(tableName).size();
+ log.debug("Waiting for {} splits to happen. Total splits: {}", expectedSplitCount,
+ actualSplitCount);
+ return actualSplitCount == expectedSplitCount;
+ });
Review Comment:
This waitFor call should not be needed. When we make an API call to add a split after the call returns the split should be there. Since all the background task completed we should expect the splits to all be there at this time, so should be able to immediately check that the expected and actual splits are equal.
--
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