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 564667112f Replace AbstractFateStore.verifyReserved() with new method
FateMutator.requireReserved() (#6521)
564667112f is described below
commit 564667112feb27a4417016e4c319d9a7c3f386ed
Author: Amanda Villarreal <[email protected]>
AuthorDate: Wed Sep 30 06:52:25 2026 -0500
Replace AbstractFateStore.verifyReserved() with new method
FateMutator.requireReserved() (#6521)
---
.../accumulo/core/fate/user/FateMutator.java | 5 +++
.../accumulo/core/fate/user/FateMutatorImpl.java | 12 ++++++
.../accumulo/core/fate/user/UserFateStore.java | 44 ++++++++--------------
3 files changed, 33 insertions(+), 28 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutator.java
b/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutator.java
index f8a6501147..52ce044511 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutator.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutator.java
@@ -52,6 +52,11 @@ public interface FateMutator<T> {
*/
FateMutator<T> requireUnreserved();
+ /**
+ * Require the transaction is reserved with the given reservation.
+ */
+ FateMutator<T> requireReserved(FateStore.FateReservation fateReservation);
+
/**
* Require the transaction has no fate key set.
*/
diff --git
a/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutatorImpl.java
b/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutatorImpl.java
index 26971e2c47..34dd4c1b67 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutatorImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/user/FateMutatorImpl.java
@@ -61,6 +61,7 @@ public class FateMutatorImpl<T> implements FateMutator<T> {
private final ConditionalMutation mutation;
private final Supplier<ConditionalWriter> writer;
private boolean requiredUnreserved = false;
+ private boolean requiredReserved = false;
public static final int INITIAL_ITERATOR_PRIO = 1000000;
public FateMutatorImpl(ClientContext context, String tableName, FateId
fateId,
@@ -108,6 +109,17 @@ public class FateMutatorImpl<T> implements FateMutator<T> {
return this;
}
+ @Override
+ public FateMutator<T> requireReserved(FateStore.FateReservation
fateReservation) {
+ Preconditions.checkState(!requiredReserved);
+ Condition condition = new
Condition(TxAdminColumnFamily.RESERVATION_COLUMN.getColumnFamily(),
+ TxAdminColumnFamily.RESERVATION_COLUMN.getColumnQualifier())
+ .setValue(fateReservation.getSerialized());
+ mutation.addCondition(condition);
+ requiredReserved = true;
+ return this;
+ }
+
@Override
public FateMutator<T> requireAbsentKey() {
Condition condition = new
Condition(TxColumnFamily.TX_KEY_COLUMN.getColumnFamily(),
diff --git
a/core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java
b/core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java
index 158c2ce1aa..dbf2421f35 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java
@@ -544,10 +544,17 @@ public class UserFateStore<T> extends
AbstractFateStore<T> {
super(fateId, reservation);
}
+ private FateMutatorImpl<T> newReservedMutator() {
+ Preconditions.checkState(isReserved(),
+ "Attempted write on unreserved FATE transaction: " + fateId);
+ Preconditions.checkState(!deleted, "Attempted write on deleted FATE
transaction: " + fateId);
+ FateMutatorImpl<T> mutator = newMutator(fateId);
+ mutator.requireReserved(reservation);
+ return mutator;
+ }
+
@Override
public Repo<T> top() {
- verifyReservedAndNotDeleted(false);
-
return scanTx(scanner -> {
scanner.setRange(getRow(fateId));
scanner.setBatchSize(1);
@@ -562,8 +569,6 @@ public class UserFateStore<T> extends AbstractFateStore<T> {
@Override
public List<ReadOnlyRepo<T>> getStack() {
- verifyReservedAndNotDeleted(false);
-
return scanTx(scanner -> {
scanner.setRange(getRow(fateId));
scanner.fetchColumnFamily(RepoColumnFamily.NAME);
@@ -577,8 +582,6 @@ public class UserFateStore<T> extends AbstractFateStore<T> {
@Override
public Serializable getTransactionInfo(TxInfo txInfo) {
- verifyReservedAndNotDeleted(false);
-
try (Scanner scanner = context.createScanner(tableName,
Authorizations.EMPTY)) {
scanner.setRange(getRow(fateId));
@@ -600,8 +603,6 @@ public class UserFateStore<T> extends AbstractFateStore<T> {
@Override
public long timeCreated() {
- verifyReservedAndNotDeleted(false);
-
return scanTx(scanner -> {
scanner.setRange(getRow(fateId));
TxColumnFamily.CREATE_TIME_COLUMN.fetch(scanner);
@@ -612,8 +613,6 @@ public class UserFateStore<T> extends AbstractFateStore<T> {
@Override
public void push(Repo<T> repo) throws StackOverflowException {
- verifyReservedAndNotDeleted(true);
-
Optional<Integer> top = findTop();
if (top.filter(t -> t >= MAX_REPOS).isPresent()) {
@@ -621,41 +620,32 @@ public class UserFateStore<T> extends
AbstractFateStore<T> {
}
FateMutator<T> fateMutator =
-
newMutator(fateId).requireStatus(REQ_PUSH_STATUS.toArray(TStatus[]::new));
+
newReservedMutator().requireStatus(REQ_PUSH_STATUS.toArray(TStatus[]::new));
fateMutator.putRepo(top.map(t -> t + 1).orElse(1), repo).mutate();
}
@Override
public void pop() {
- verifyReservedAndNotDeleted(true);
-
Optional<Integer> top = findTop();
- top.ifPresent(t ->
newMutator(fateId).requireStatus(REQ_POP_STATUS.toArray(TStatus[]::new))
- .deleteRepo(t).mutate());
+
newReservedMutator().requireStatus(REQ_POP_STATUS.toArray(TStatus[]::new))
+ .deleteRepo(top.orElseThrow()).mutate();
}
@Override
public void setStatus(TStatus status) {
- verifyReservedAndNotDeleted(true);
-
- newMutator(fateId).putStatus(status).mutate();
+ newReservedMutator().putStatus(status).mutate();
observedStatus = status;
}
@Override
public void setTransactionInfo(TxInfo txInfo, Serializable so) {
- verifyReservedAndNotDeleted(true);
-
final byte[] serialized = serializeTxInfo(so);
-
- newMutator(fateId).putTxInfo(txInfo, serialized).mutate();
+ newReservedMutator().putTxInfo(txInfo, serialized).mutate();
}
@Override
public void delete() {
- verifyReservedAndNotDeleted(true);
-
- var mutator = newMutator(fateId);
+ var mutator = newReservedMutator();
mutator.requireStatus(REQ_DELETE_STATUS.toArray(TStatus[]::new));
mutator.delete().mutate();
this.deleted = true;
@@ -663,9 +653,7 @@ public class UserFateStore<T> extends AbstractFateStore<T> {
@Override
public void forceDelete() {
- verifyReservedAndNotDeleted(true);
-
- var mutator = newMutator(fateId);
+ var mutator = newReservedMutator();
mutator.requireStatus(REQ_FORCE_DELETE_STATUS.toArray(TStatus[]::new));
mutator.delete().mutate();
this.deleted = true;
