You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2023/05/26 23:09:14 UTC
[accumulo] branch elasticity updated: adds validation to user split and new tests (#3431)
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 5608a9291a adds validation to user split and new tests (#3431)
5608a9291a is described below
commit 5608a9291a38b449c5dd7b1084a84719cf7c0294
Author: Keith Turner <kt...@apache.org>
AuthorDate: Fri May 26 19:09:08 2023 -0400
adds validation to user split and new tests (#3431)
---
.../accumulo/manager/FateServiceHandler.java | 31 +++++++++++++++++
.../apache/accumulo/test/functional/SplitIT.java | 39 +++++++++++++++++++++-
2 files changed, 69 insertions(+), 1 deletion(-)
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java b/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java
index 9cff4251dc..7b13303d70 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java
@@ -43,6 +43,7 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.apache.accumulo.core.client.AccumuloSecurityException;
@@ -75,6 +76,7 @@ import org.apache.accumulo.core.manager.thrift.ThriftPropertyException;
import org.apache.accumulo.core.securityImpl.thrift.TCredentials;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.FastFormat;
+import org.apache.accumulo.core.util.TextUtil;
import org.apache.accumulo.core.util.Validator;
import org.apache.accumulo.core.util.tables.TableNameUtil;
import org.apache.accumulo.core.volume.Volume;
@@ -740,6 +742,35 @@ class FateServiceHandler implements FateService.Iface {
.map(ByteBufferUtil::toText).collect(Collectors.toCollection(TreeSet::new));
KeyExtent extent = new KeyExtent(tableId, endRow, prevEndRow);
+
+ Predicate<Text> outOfBoundsTest =
+ split -> !extent.contains(split) || split.equals(extent.endRow());
+
+ if (splits.stream().anyMatch(outOfBoundsTest)) {
+ splits.stream().filter(outOfBoundsTest).forEach(split -> log
+ .warn("split for {} is out of bounds : {}", extent, TextUtil.truncate(split)));
+
+ throw new ThriftTableOperationException(tableId.canonical(), null, tableOp,
+ TableOperationExceptionType.OTHER,
+ "Split is outside bounds of tablet or equal to the tablets endrow, see warning in logs for more information.");
+ }
+
+ var maxSplitSize = manager.getContext().getTableConfiguration(tableId)
+ .getAsBytes(Property.TABLE_MAX_END_ROW_SIZE);
+
+ Predicate<Text> oversizedTest = split -> split.getLength() > maxSplitSize;
+
+ if (splits.stream().anyMatch(oversizedTest)) {
+ splits.stream().filter(oversizedTest)
+ .forEach(split -> log.warn(
+ "split exceeds max configured split size len:{} max:{} extent:{} split:{}",
+ split.getLength(), maxSplitSize, extent, TextUtil.truncate(split)));
+
+ throw new ThriftTableOperationException(tableId.canonical(), null, tableOp,
+ TableOperationExceptionType.OTHER,
+ "Length of requested split exceeds tables configured max, see warning in logs for more information.");
+ }
+
manager.requestUnassignment(extent, opid);
goalMessage = "Splitting " + extent + " for user into " + (splits.size() + 1) + " tablets";
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java b/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java
index f143c91480..3a342b78e8 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java
@@ -20,16 +20,20 @@ package org.apache.accumulo.test.functional;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
import java.time.Duration;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.TreeSet;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.admin.InstanceOperations;
import org.apache.accumulo.core.client.admin.NewTableConfiguration;
@@ -50,6 +54,7 @@ import org.apache.accumulo.test.TestIngest;
import org.apache.accumulo.test.VerifyIngest;
import org.apache.accumulo.test.VerifyIngest.VerifyParams;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -154,7 +159,7 @@ public class SplitIT extends AccumuloClusterHarness {
}
assertTrue(shortened > 0, "Shortened should be greater than zero: " + shortened);
- assertTrue(count > 10, "Count should be cgreater than 10: " + count);
+ assertTrue(count > 10, "Count should be greater than 10: " + count);
}
assertEquals(0, getCluster().getClusterControl().exec(CheckForMetadataProblems.class,
@@ -205,4 +210,36 @@ public class SplitIT extends AccumuloClusterHarness {
}
}
+ @Test
+ public void testLargeSplit() throws Exception {
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ String tableName = getUniqueNames(1)[0];
+ c.tableOperations().create(tableName, new NewTableConfiguration()
+ .setProperties(Map.of(Property.TABLE_MAX_END_ROW_SIZE.getKey(), "10K")));
+
+ byte[] okSplit = new byte[4096];
+ for (int i = 0; i < okSplit.length; i++) {
+ okSplit[i] = (byte) (i % 256);
+ }
+
+ var splits1 = new TreeSet<Text>(List.of(new Text(okSplit)));
+
+ c.tableOperations().addSplits(tableName, splits1);
+
+ assertEquals(splits1, new TreeSet<>(c.tableOperations().listSplits(tableName)));
+
+ byte[] bigSplit = new byte[4096 * 4];
+ for (int i = 0; i < bigSplit.length; i++) {
+ bigSplit[i] = (byte) (i % 256);
+ }
+
+ var splits2 = new TreeSet<Text>(List.of(new Text(bigSplit)));
+ // split should fail because it exceeds the configured max split size
+ assertThrows(AccumuloException.class,
+ () -> c.tableOperations().addSplits(tableName, splits2));
+
+ // ensure the large split is not there
+ assertEquals(splits1, new TreeSet<>(c.tableOperations().listSplits(tableName)));
+ }
+ }
}