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];