This is an automated email from the ASF dual-hosted git repository.
ddanielr pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new 01f48adf38 Remove active FateTx constraint for bulk import (#4063)
01f48adf38 is described below
commit 01f48adf385faebb3bc2e5f3fb4e69dd4af12e3f
Author: Daniel Roberts <ddani...@gmail.com>
AuthorDate: Wed Dec 13 09:58:13 2023 -0500
Remove active FateTx constraint for bulk import (#4063)
* Remove active FateTx constraint for bulk import
Removes the active FateTx ID constraint for metadata mutations when
creating bulk import file markers.
Changes the MutationsRejectedException to return the violation type
codes.
---------
Co-authored-by: Keith Turner <ktur...@apache.org>
---
.../core/client/MutationsRejectedException.java | 18 +++--
.../server/constraints/MetadataConstraints.java | 22 +-----
.../constraints/MetadataConstraintsTest.java | 84 ++++------------------
3 files changed, 30 insertions(+), 94 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/client/MutationsRejectedException.java
b/core/src/main/java/org/apache/accumulo/core/client/MutationsRejectedException.java
index e3f773364c..30e6fbd183 100644
---
a/core/src/main/java/org/apache/accumulo/core/client/MutationsRejectedException.java
+++
b/core/src/main/java/org/apache/accumulo/core/client/MutationsRejectedException.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.accumulo.core.client.security.SecurityErrorCode;
import org.apache.accumulo.core.clientImpl.ClientContext;
@@ -59,8 +60,11 @@ public class MutationsRejectedException extends
AccumuloException {
Map<TabletId,Set<SecurityErrorCode>> hashMap, Collection<String>
serverSideErrors,
int unknownErrors, Throwable cause) {
super(
- "# constraint violations : " + cvsList.size() + " security codes: " +
hashMap.toString()
- + " # server errors " + serverSideErrors.size() + " # exceptions
" + unknownErrors,
+ "constraint violation codes : "
+ +
cvsList.stream().map(ConstraintViolationSummary::getViolationCode)
+ .collect(Collectors.toSet())
+ + " security codes: " + hashMap.toString() + " # server errors "
+ + serverSideErrors.size() + " # exceptions " + unknownErrors,
cause);
this.cvsl.addAll(cvsList);
this.af.putAll(hashMap);
@@ -82,9 +86,13 @@ public class MutationsRejectedException extends
AccumuloException {
public MutationsRejectedException(AccumuloClient client,
List<ConstraintViolationSummary> cvsList,
Map<TabletId,Set<SecurityErrorCode>> hashMap, Collection<String>
serverSideErrors,
int unknownErrors, Throwable cause) {
- super("# constraint violations : " + cvsList.size() + " security codes: "
- + format(hashMap, (ClientContext) client) + " # server errors " +
serverSideErrors.size()
- + " # exceptions " + unknownErrors, cause);
+ super(
+ "constraint violation codes : "
+ +
cvsList.stream().map(ConstraintViolationSummary::getViolationCode).collect(
+ Collectors.toSet())
+ + " security codes: " + format(hashMap, (ClientContext) client) +
" # server errors "
+ + serverSideErrors.size() + " # exceptions " + unknownErrors,
+ cause);
this.cvsl.addAll(cvsList);
this.af.putAll(hashMap);
this.es.addAll(serverSideErrors);
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
b/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
index 1b5a074adb..ced97d0a68 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
@@ -24,10 +24,8 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
-import java.util.Objects;
import java.util.Set;
-import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.data.ColumnUpdate;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -54,8 +52,6 @@ import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Ta
import org.apache.accumulo.core.util.ColumnFQ;
import org.apache.accumulo.core.util.cleaner.CleanerUtil;
import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.server.zookeeper.TransactionWatcher.Arbitrator;
-import org.apache.accumulo.server.zookeeper.TransactionWatcher.ZooArbitrator;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -212,7 +208,7 @@ public class MetadataConstraints implements Constraint {
violations = addViolation(violations, 1);
}
} else if (columnFamily.equals(ScanFileColumnFamily.NAME)) {
-
+ // Do nothing if ScanFile ref
} else if (columnFamily.equals(BulkFileColumnFamily.NAME)) {
if (!columnUpdate.isDeleted() && !checkedBulk) {
// splits, which also write the time reference, are allowed to write
this reference even
@@ -252,14 +248,7 @@ public class MetadataConstraints implements Constraint {
}
if (!isSplitMutation && !isLocationMutation) {
- long tid = BulkFileColumnFamily.getBulkLoadTid(new
Value(tidString));
-
- try {
- if (otherTidCount > 0 || !dataFiles.equals(loadedFiles) ||
!getArbitrator(context)
- .transactionAlive(Constants.BULK_ARBITRATOR_TYPE, tid)) {
- violations = addViolation(violations, 8);
- }
- } catch (Exception ex) {
+ if (otherTidCount > 0 || !dataFiles.equals(loadedFiles)) {
violations = addViolation(violations, 8);
}
}
@@ -320,11 +309,6 @@ public class MetadataConstraints implements Constraint {
return violations;
}
- protected Arbitrator getArbitrator(ServerContext context) {
- Objects.requireNonNull(context);
- return new ZooArbitrator(context);
- }
-
@Override
public String getViolationDescription(short violationCode) {
switch (violationCode) {
@@ -343,7 +327,7 @@ public class MetadataConstraints implements Constraint {
case 7:
return "Lock not held in zookeeper by writer";
case 8:
- return "Bulk load transaction no longer running";
+ return "Bulk load mutation contains either inconsistent files or
multiple fateTX ids";
}
return null;
}
diff --git
a/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java
b/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java
index 6128096ecb..bd59fef1e7 100644
---
a/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java
+++
b/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java
@@ -33,34 +33,12 @@ import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Da
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily;
import
org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.server.zookeeper.TransactionWatcher.Arbitrator;
import org.apache.hadoop.io.Text;
import org.easymock.EasyMock;
import org.junit.jupiter.api.Test;
public class MetadataConstraintsTest {
- static class TestMetadataConstraints extends MetadataConstraints {
- @Override
- protected Arbitrator getArbitrator(ServerContext context) {
- return new Arbitrator() {
-
- @Override
- public boolean transactionAlive(String type, long tid) {
- if (tid == 9) {
- throw new RuntimeException("txid 9 reserved for future use");
- }
- return tid == 5 || tid == 7;
- }
-
- @Override
- public boolean transactionComplete(String type, long tid) {
- return tid != 5 && tid != 7;
- }
- };
- }
- }
-
private SystemEnvironment createEnv() {
SystemEnvironment env = EasyMock.createMock(SystemEnvironment.class);
ServerContext context = EasyMock.createMock(ServerContext.class);
@@ -146,31 +124,11 @@ public class MetadataConstraintsTest {
@Test
public void testBulkFileCheck() {
- MetadataConstraints mc = new TestMetadataConstraints();
+ MetadataConstraints mc = new MetadataConstraints();
Mutation m;
List<Short> violations;
- // inactive txid
- m = new Mutation(new Text("0;foo"));
- m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new
Value("12345"));
- m.put(DataFileColumnFamily.NAME, new Text("/someFile"),
- new DataFileValue(1, 1).encodeAsValue());
- violations = mc.check(createEnv(), m);
- assertNotNull(violations);
- assertEquals(1, violations.size());
- assertEquals(Short.valueOf((short) 8), violations.get(0));
-
- // txid that throws exception
- m = new Mutation(new Text("0;foo"));
- m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("9"));
- m.put(DataFileColumnFamily.NAME, new Text("/someFile"),
- new DataFileValue(1, 1).encodeAsValue());
- violations = mc.check(createEnv(), m);
- assertNotNull(violations);
- assertEquals(1, violations.size());
- assertEquals(Short.valueOf((short) 8), violations.get(0));
-
- // active txid w/ file
+ // loaded marker w/ file
m = new Mutation(new Text("0;foo"));
m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
m.put(DataFileColumnFamily.NAME, new Text("/someFile"),
@@ -178,7 +136,7 @@ public class MetadataConstraintsTest {
violations = mc.check(createEnv(), m);
assertNull(violations);
- // active txid w/o file
+ // loaded marker w/o file
m = new Mutation(new Text("0;foo"));
m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
violations = mc.check(createEnv(), m);
@@ -186,31 +144,31 @@ public class MetadataConstraintsTest {
assertEquals(1, violations.size());
assertEquals(Short.valueOf((short) 8), violations.get(0));
- // two active txids w/ files
+ // two files w/ same txid
m = new Mutation(new Text("0;foo"));
m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
m.put(DataFileColumnFamily.NAME, new Text("/someFile"),
new DataFileValue(1, 1).encodeAsValue());
- m.put(BulkFileColumnFamily.NAME, new Text("/someFile2"), new Value("7"));
+ m.put(BulkFileColumnFamily.NAME, new Text("/someFile2"), new Value("5"));
m.put(DataFileColumnFamily.NAME, new Text("/someFile2"),
new DataFileValue(1, 1).encodeAsValue());
violations = mc.check(createEnv(), m);
- assertNotNull(violations);
- assertEquals(1, violations.size());
- assertEquals(Short.valueOf((short) 8), violations.get(0));
+ assertNull(violations);
- // two files w/ one active txid
+ // two files w/ different txid
m = new Mutation(new Text("0;foo"));
m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
m.put(DataFileColumnFamily.NAME, new Text("/someFile"),
new DataFileValue(1, 1).encodeAsValue());
- m.put(BulkFileColumnFamily.NAME, new Text("/someFile2"), new Value("5"));
+ m.put(BulkFileColumnFamily.NAME, new Text("/someFile2"), new Value("7"));
m.put(DataFileColumnFamily.NAME, new Text("/someFile2"),
new DataFileValue(1, 1).encodeAsValue());
violations = mc.check(createEnv(), m);
- assertNull(violations);
+ assertNotNull(violations);
+ assertEquals(1, violations.size());
+ assertEquals(Short.valueOf((short) 8), violations.get(0));
- // two loaded w/ one active txid and one file
+ // two loaded markers but only one file.
m = new Mutation(new Text("0;foo"));
m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
m.put(DataFileColumnFamily.NAME, new Text("/someFile"),
@@ -221,34 +179,20 @@ public class MetadataConstraintsTest {
assertEquals(1, violations.size());
assertEquals(Short.valueOf((short) 8), violations.get(0));
- // active txid, mutation that looks like split
+ // mutation that looks like split
m = new Mutation(new Text("0;foo"));
m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value("/t1"));
violations = mc.check(createEnv(), m);
assertNull(violations);
- // inactive txid, mutation that looks like split
- m = new Mutation(new Text("0;foo"));
- m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new
Value("12345"));
- ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value("/t1"));
- violations = mc.check(createEnv(), m);
- assertNull(violations);
-
- // active txid, mutation that looks like a load
+ // mutation that looks like a load
m = new Mutation(new Text("0;foo"));
m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new Value("5"));
m.put(CurrentLocationColumnFamily.NAME, new Text("789"), new
Value("127.0.0.1:9997"));
violations = mc.check(createEnv(), m);
assertNull(violations);
- // inactive txid, mutation that looks like a load
- m = new Mutation(new Text("0;foo"));
- m.put(BulkFileColumnFamily.NAME, new Text("/someFile"), new
Value("12345"));
- m.put(CurrentLocationColumnFamily.NAME, new Text("789"), new
Value("127.0.0.1:9997"));
- violations = mc.check(createEnv(), m);
- assertNull(violations);
-
// deleting a load flag
m = new Mutation(new Text("0;foo"));
m.putDelete(BulkFileColumnFamily.NAME, new Text("/someFile"));
