This is an automated email from the ASF dual-hosted git repository.
cshannon 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 d9d6121684 Make sure AccumuloStore persists ageoff txinfo (#4199)
d9d6121684 is described below
commit d9d61216842903e1de3533eda760c2c27afbca3a
Author: Christopher L. Shannon <cshan...@apache.org>
AuthorDate: Fri Jan 26 14:25:13 2024 -0500
Make sure AccumuloStore persists ageoff txinfo (#4199)
The changes in #4169 moved age off tracking out of memory and into the
FATE store. The AccumuloStore has a bug where it doesn't handle the new
TxInfo enum type of TX_AGEOFF so the info is never persisted to the
store. This fixes the store so it will correctly read/write the value
and also adds a default enum case to throw an exception in the future in
case a new value is passed that is unknown so it will be caught. A test
was also added to iterate over all types to verify they all work for the
AccumuloStore so if a new info type is ever added it will validate it
in the future.
Co-authored-by: Keith Turner <ktur...@apache.org>
---
.../accumulo/core/fate/accumulo/AccumuloStore.java | 21 ++++-------------
.../accumulo/core/fate/accumulo/FateMutator.java | 2 ++
.../core/fate/accumulo/FateMutatorImpl.java | 11 +++++++++
.../core/fate/accumulo/schema/FateSchema.java | 3 +++
.../fate/accumulo/AccumuloStoreReadWriteIT.java | 27 ++++++++++++++++++++++
5 files changed, 47 insertions(+), 17 deletions(-)
diff --git
a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/AccumuloStore.java
b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/AccumuloStore.java
index a31e081e7e..b5dc999d42 100644
---
a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/AccumuloStore.java
+++
b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/AccumuloStore.java
@@ -188,6 +188,9 @@ public class AccumuloStore<T> extends AbstractFateStore<T> {
case RETURN_VALUE:
cq = TxInfoColumnFamily.RETURN_VALUE_COLUMN;
break;
+ case TX_AGEOFF:
+ cq = TxInfoColumnFamily.TX_AGEOFF_COLUMN;
+ break;
default:
throw new IllegalArgumentException("Unexpected TxInfo type " +
txInfo);
}
@@ -248,23 +251,7 @@ public class AccumuloStore<T> extends AbstractFateStore<T>
{
FateMutator<T> fateMutator = newMutator(tid);
final byte[] serialized = serializeTxInfo(so);
-
- switch (txInfo) {
- case TX_NAME:
- fateMutator.putName(serialized);
- break;
- case AUTO_CLEAN:
- fateMutator.putAutoClean(serialized);
- break;
- case EXCEPTION:
- fateMutator.putException(serialized);
- break;
- case RETURN_VALUE:
- fateMutator.putReturnValue(serialized);
- break;
- default:
- throw new IllegalArgumentException("Unexpected TxInfo type " +
txInfo);
- }
+ fateMutator.putTxInfo(txInfo, serialized);
fateMutator.mutate();
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutator.java
b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutator.java
index 306841612e..4caf5985bd 100644
--- a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutator.java
+++ b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutator.java
@@ -36,6 +36,8 @@ public interface FateMutator<T> {
FateMutator<T> putReturnValue(byte[] data);
+ FateMutator<T> putAgeOff(byte[] data);
+
FateMutator<T> putTxInfo(Fate.TxInfo txInfo, byte[] data);
FateMutator<T> putRepo(int position, Repo<T> repo);
diff --git
a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutatorImpl.java
b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutatorImpl.java
index b605b91097..90d22008d5 100644
---
a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutatorImpl.java
+++
b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/FateMutatorImpl.java
@@ -90,6 +90,12 @@ public class FateMutatorImpl<T> implements FateMutator<T> {
return this;
}
+ @Override
+ public FateMutator<T> putAgeOff(byte[] data) {
+ TxInfoColumnFamily.TX_AGEOFF_COLUMN.put(mutation, new Value(data));
+ return this;
+ }
+
@Override
public FateMutator<T> putTxInfo(TxInfo txInfo, byte[] data) {
switch (txInfo) {
@@ -105,6 +111,11 @@ public class FateMutatorImpl<T> implements FateMutator<T> {
case RETURN_VALUE:
putReturnValue(data);
break;
+ case TX_AGEOFF:
+ putAgeOff(data);
+ break;
+ default:
+ throw new IllegalArgumentException("Unexpected TxInfo type: " +
txInfo);
}
return this;
}
diff --git
a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/schema/FateSchema.java
b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/schema/FateSchema.java
index dbb84049a8..7e4e639a7c 100644
---
a/core/src/main/java/org/apache/accumulo/core/fate/accumulo/schema/FateSchema.java
+++
b/core/src/main/java/org/apache/accumulo/core/fate/accumulo/schema/FateSchema.java
@@ -49,6 +49,9 @@ public class FateSchema {
public static final String RETURN_VALUE = "retval";
public static final ColumnFQ RETURN_VALUE_COLUMN = new ColumnFQ(NAME, new
Text(RETURN_VALUE));
+
+ public static final String TX_AGEOFF = "txageoff";
+ public static final ColumnFQ TX_AGEOFF_COLUMN = new ColumnFQ(NAME, new
Text(TX_AGEOFF));
}
public static class RepoColumnFamily {
diff --git
a/test/src/main/java/org/apache/accumulo/test/fate/accumulo/AccumuloStoreReadWriteIT.java
b/test/src/main/java/org/apache/accumulo/test/fate/accumulo/AccumuloStoreReadWriteIT.java
index 0d36d30a95..236cb9624c 100644
---
a/test/src/main/java/org/apache/accumulo/test/fate/accumulo/AccumuloStoreReadWriteIT.java
+++
b/test/src/main/java/org/apache/accumulo/test/fate/accumulo/AccumuloStoreReadWriteIT.java
@@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Duration;
@@ -127,6 +128,32 @@ public class AccumuloStoreReadWriteIT extends
SharedMiniClusterBase {
}
}
+ @Test
+ public void testReadWriteTxInfo() throws Exception {
+ final String table = getUniqueNames(1)[0];
+ try (ClientContext client =
+ (ClientContext) Accumulo.newClient().from(getClientProps()).build()) {
+ client.tableOperations().create(table);
+
+ AccumuloStore<TestEnv> store = new AccumuloStore<>(client, table);
+
+ long tid = store.create();
+ FateTxStore<TestEnv> txStore = store.reserve(tid);
+
+ try {
+ // Go through all enum values to verify each TxInfo type will be
properly
+ // written and read from the store
+ for (TxInfo txInfo : TxInfo.values()) {
+ assertNull(txStore.getTransactionInfo(txInfo));
+ txStore.setTransactionInfo(txInfo, "value: " + txInfo.name());
+ assertEquals("value: " + txInfo.name(),
txStore.getTransactionInfo(txInfo));
+ }
+ } finally {
+ txStore.delete();
+ }
+ }
+ }
+
@Test
public void testDeferredOverflow() throws Exception {
final String table = getUniqueNames(1)[0];
