This is an automated email from the ASF dual-hosted git repository.
ivandika3 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 8bdfef0b8fb HDDS-13365. OM support automatic table cache tracking and
cleanup (#10171)
8bdfef0b8fb is described below
commit 8bdfef0b8fb21a5d6db4528fede53abb13b150f1
Author: Ivan Andika <[email protected]>
AuthorDate: Tue Sep 29 15:15:17 2026 +0800
HDDS-13365. OM support automatic table cache tracking and cleanup (#10171)
Co-authored-by: Tsz-Wo Nicholas Sze <[email protected]>
---
.../apache/hadoop/hdds/utils/db/TypedTable.java | 2 +
.../utils/db/cache/TableCacheUpdateTracker.java | 88 +++++++++++++++
.../hdds/utils/db/TestTableCacheUpdateTracker.java | 118 +++++++++++++++++++++
.../ozone/om/ratis/OzoneManagerDoubleBuffer.java | 30 +++---
.../hadoop/ozone/om/request/OMClientRequest.java | 7 +-
.../hadoop/ozone/om/response/OMClientResponse.java | 24 ++++-
.../om/response/key/OMOpenKeysDeleteResponse.java | 8 --
.../protocolPB/OzoneManagerRequestHandler.java | 4 +-
...tOzoneManagerDoubleBufferWithDummyResponse.java | 93 ++++++++++++++--
.../request/key/TestOMOpenKeysDeleteRequest.java | 7 ++
.../ozone/om/response/TestCleanupTableInfo.java | 3 +
.../ozone/om/response/TestOMClientResponse.java | 51 +++++++++
12 files changed, 398 insertions(+), 37 deletions(-)
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java
index b9c0c56feba..cc5d661cd03 100644
---
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java
@@ -40,6 +40,7 @@
import org.apache.hadoop.hdds.utils.db.cache.PartialTableCache;
import org.apache.hadoop.hdds.utils.db.cache.TableCache;
import org.apache.hadoop.hdds.utils.db.cache.TableCache.CacheType;
+import org.apache.hadoop.hdds.utils.db.cache.TableCacheUpdateTracker;
import org.apache.hadoop.hdds.utils.db.cache.TableNoCache;
import org.apache.ratis.util.Preconditions;
import org.apache.ratis.util.function.CheckedBiFunction;
@@ -587,6 +588,7 @@ public long getEstimatedKeyCount() throws
RocksDatabaseException {
public void addCacheEntry(CacheKey<KEY> cacheKey,
CacheValue<VALUE> cacheValue) {
// This will override the entry if there is already entry for this key.
+ TableCacheUpdateTracker.recordCacheUpdate(getName());
cache.put(cacheKey, cacheValue);
}
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/cache/TableCacheUpdateTracker.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/cache/TableCacheUpdateTracker.java
new file mode 100644
index 00000000000..0bb5413d3e4
--- /dev/null
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/cache/TableCacheUpdateTracker.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.utils.db.cache;
+
+import java.util.LinkedHashSet;
+import java.util.Objects;
+import java.util.Set;
+import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.util.UncheckedAutoCloseable;
+
+/**
+ * Tracks table caches updated by the current thread.
+ * Only one tracker is allowed to be active per thread.
+ * <p>
+ * Only one thread, which is the thread invoked the constructor, can access
the non-static methods.
+ * The call sequences must be:
+ * {@link #TableCacheUpdateTracker()},
+ * {@link #record(String)} (any number of times including zero),
+ * {@link #removeUpdatedTables()} (exactly one time),
+ * {@link #close()} (exactly one time).
+ */
+public final class TableCacheUpdateTracker implements UncheckedAutoCloseable {
+ private static final ThreadLocal<TableCacheUpdateTracker> CURRENT = new
ThreadLocal<>();
+
+ private final Thread thread = Thread.currentThread();
+ private Set<String> tables = null;
+
+ public static TableCacheUpdateTracker track() {
+ TableCacheUpdateTracker tracker = new TableCacheUpdateTracker();
+ CURRENT.set(tracker);
+ return tracker;
+ }
+
+ public static void recordCacheUpdate(String tableName) {
+ TableCacheUpdateTracker tracker = CURRENT.get();
+ if (tracker != null) {
+ tracker.record(tableName);
+ }
+ }
+
+ private TableCacheUpdateTracker() {
+ }
+
+ private void assertCurrent() {
+ Preconditions.assertSame(this, CURRENT.get(), "tracker");
+ Preconditions.assertSame(thread, Thread.currentThread(), "thread");
+ }
+
+ public Set<String> removeUpdatedTables() {
+ assertCurrent();
+ final Set<String> t = tables;
+ tables = null;
+ return t; // can be null
+ }
+
+ @Override
+ public void close() {
+ assertCurrent(); // not idempotent
+ Preconditions.assertNull(tables, "tables");
+ CURRENT.remove();
+ }
+
+ private void record(String tableName) {
+ assertCurrent();
+ Objects.requireNonNull(tableName, "tableName");
+ Preconditions.assertTrue(!tableName.isEmpty(), "tableName is empty");
+
+ if (tables == null) {
+ tables = new LinkedHashSet<>();
+ }
+ tables.add(tableName);
+ }
+}
diff --git
a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/TestTableCacheUpdateTracker.java
b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/TestTableCacheUpdateTracker.java
new file mode 100644
index 00000000000..adf4416b9ad
--- /dev/null
+++
b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/TestTableCacheUpdateTracker.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.utils.db;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.assertj.core.api.Assertions.assertThatNullPointerException;
+
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hadoop.hdds.utils.db.cache.TableCacheUpdateTracker;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests {@link TableCacheUpdateTracker}.
+ */
+public class TestTableCacheUpdateTracker {
+
+ @Test
+ public void trackReturnsScopedTrackerWithUpdatedTables() {
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
+ TableCacheUpdateTracker.recordCacheUpdate("table1");
+ TableCacheUpdateTracker.recordCacheUpdate("table2");
+
+ assertThat(tracker.removeUpdatedTables())
+ .containsExactly("table1", "table2");
+ }
+ }
+
+ @Test
+ public void closeRequiresUpdatedTablesToBeRemoved() {
+ TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track();
+ TableCacheUpdateTracker.recordCacheUpdate("table1");
+
+ assertThatIllegalStateException().isThrownBy(tracker::close);
+
+ assertThat(tracker.removeUpdatedTables()).containsExactly("table1");
+ tracker.close();
+ }
+
+ @Test
+ public void removeUpdatedTablesReturnsNullWithoutUpdates() {
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
+ assertThat(tracker.removeUpdatedTables()).isNull();
+ }
+ }
+
+ @Test
+ public void recordCacheUpdateDeduplicatesTablesInInsertionOrder() {
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
+ TableCacheUpdateTracker.recordCacheUpdate("table1");
+ TableCacheUpdateTracker.recordCacheUpdate("table2");
+ TableCacheUpdateTracker.recordCacheUpdate("table1");
+
+ assertThat(tracker.removeUpdatedTables())
+ .containsExactly("table1", "table2");
+ }
+ }
+
+ @Test
+ public void recordCacheUpdateRejectsInvalidTableNames() {
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
+ assertThatNullPointerException()
+ .isThrownBy(() -> TableCacheUpdateTracker.recordCacheUpdate(null));
+ assertThatIllegalStateException()
+ .isThrownBy(() -> TableCacheUpdateTracker.recordCacheUpdate(""));
+
+ assertThat(tracker.removeUpdatedTables()).isNull();
+ }
+ }
+
+ @Test
+ public void updatesDoNotLeakIntoSubsequentTracker() {
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
+ TableCacheUpdateTracker.recordCacheUpdate("table1");
+ assertThat(tracker.removeUpdatedTables()).containsExactly("table1");
+ }
+
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
+ TableCacheUpdateTracker.recordCacheUpdate("table2");
+ assertThat(tracker.removeUpdatedTables()).containsExactly("table2");
+ }
+ }
+
+ @Test
+ public void nonOwnerThreadCannotAccessTracker() throws InterruptedException {
+ TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread thread = new Thread(() -> {
+ try {
+ tracker.removeUpdatedTables();
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+
+ thread.start();
+ thread.join();
+
+ assertThat(failure.get()).isInstanceOf(IllegalStateException.class);
+ assertThat(tracker.removeUpdatedTables()).isNull();
+ tracker.close();
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
index c822e0b9fe2..5de26d73ad4 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
@@ -528,26 +528,20 @@ private static boolean
isStandaloneBatchCmdTypes(OMResponse response) {
}
private void addCleanupEntry(Entry entry, Map<String, List<Long>>
cleanupEpochs) {
- Class<? extends OMClientResponse> responseClass =
- entry.getResponse().getClass();
+ OMClientResponse response = entry.getResponse();
CleanupTableInfo cleanupTableInfo =
- responseClass.getAnnotation(CleanupTableInfo.class);
- if (cleanupTableInfo != null) {
- final List<String> cleanupTables;
- if (cleanupTableInfo.cleanupAll()) {
- cleanupTables = OMDBDefinition.get().getColumnFamilyNames();
- } else {
- cleanupTables = Arrays.asList(cleanupTableInfo.cleanupTables());
- }
- for (String table : cleanupTables) {
- cleanupEpochs.computeIfAbsent(table, list -> new ArrayList<>())
- .add(entry.getTermIndex().getIndex());
- }
+ response.getClass().getAnnotation(CleanupTableInfo.class);
+ final Iterable<String> cleanupTables;
+ if (cleanupTableInfo == null) {
+ cleanupTables = response.removeCleanupTables();
+ } else if (cleanupTableInfo.cleanupAll()) {
+ cleanupTables = OMDBDefinition.get().getColumnFamilyNames();
} else {
- // This is to catch early errors, when a new response class missed to
- // add CleanupTableInfo annotation.
- throw new RuntimeException("CleanupTableInfo Annotation is missing " +
- "for" + responseClass);
+ cleanupTables = Arrays.asList(cleanupTableInfo.cleanupTables());
+ }
+ for (String table : cleanupTables) {
+ cleanupEpochs.computeIfAbsent(table, list -> new ArrayList<>())
+ .add(entry.getTermIndex().getIndex());
}
}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
index b6cca710d78..62013f7c071 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java
@@ -29,6 +29,7 @@
import java.util.Objects;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.hdds.utils.db.cache.TableCacheUpdateTracker;
import org.apache.hadoop.ipc_.ProtobufRpcEngine;
import org.apache.hadoop.ozone.OmUtils;
import org.apache.hadoop.ozone.OzoneConsts;
@@ -165,7 +166,11 @@ public void handleRequestFailure(OzoneManager
ozoneManager) {
public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
long transactionLogIndex) {
ExecutionContext context = ExecutionContext.of(transactionLogIndex,
TransactionInfo.getTermIndex(transactionLogIndex));
- return validateAndUpdateCache(ozoneManager, context);
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
+ OMClientResponse response = validateAndUpdateCache(ozoneManager,
context);
+ response.addCleanupTables(tracker.removeUpdatedTables());
+ return response;
+ }
}
@VisibleForTesting
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/OMClientResponse.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/OMClientResponse.java
index ffe3f70099c..8ca2189bc9b 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/OMClientResponse.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/OMClientResponse.java
@@ -19,7 +19,10 @@
import com.google.common.base.Preconditions;
import java.io.IOException;
+import java.util.Collections;
import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.hadoop.hdds.utils.db.BatchOperation;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
@@ -34,6 +37,7 @@ public abstract class OMClientResponse {
private final OMResponse omResponse;
private OMLockDetails omLockDetails;
+ private final AtomicReference<Set<String>> cleanupTables = new
AtomicReference<>();
public OMClientResponse(OMResponse omResponse) {
Objects.requireNonNull(omResponse, "omResponse == null");
@@ -86,5 +90,23 @@ public void setOmLockDetails(
OMLockDetails omLockDetails) {
this.omLockDetails = omLockDetails;
}
-}
+ public Set<String> removeCleanupTables() {
+ final Set<String> tables = cleanupTables.getAndSet(null);
+ return tables == null ? Collections.emptySet() :
Collections.unmodifiableSet(tables);
+ }
+
+ public void addCleanupTables(Set<String> tables) {
+ if (tables == null || tables.isEmpty()) {
+ return;
+ }
+ cleanupTables.updateAndGet(t -> {
+ if (t == null || t.isEmpty()) {
+ return tables;
+ } else {
+ t.addAll(tables);
+ return t;
+ }
+ });
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMOpenKeysDeleteResponse.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMOpenKeysDeleteResponse.java
index e46bd7db64f..e507159894e 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMOpenKeysDeleteResponse.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMOpenKeysDeleteResponse.java
@@ -17,11 +17,6 @@
package org.apache.hadoop.ozone.om.response.key;
-import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.BUCKET_TABLE;
-import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.DELETED_TABLE;
-import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_FILE_TABLE;
-import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_KEY_TABLE;
-
import jakarta.annotation.Nonnull;
import java.io.IOException;
import java.util.Map;
@@ -31,15 +26,12 @@
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
-import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
/**
* Handles responses to move open keys from the open key table to the delete
* table. Modifies the open key table and delete table databases.
*/
-@CleanupTableInfo(cleanupTables = {OPEN_KEY_TABLE, OPEN_FILE_TABLE,
- DELETED_TABLE, BUCKET_TABLE})
public class OMOpenKeysDeleteResponse extends AbstractOMKeyDeleteResponse {
private Map<String, Pair<Long, OmKeyInfo>> keysToDelete;
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
index 327dd9d53dc..2fc6000abe6 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java
@@ -60,6 +60,7 @@
import
org.apache.hadoop.hdds.protocol.proto.HddsProtos.UpgradeFinalizationStatus;
import org.apache.hadoop.hdds.scm.protocolPB.OzonePBHelper;
import org.apache.hadoop.hdds.utils.FaultInjector;
+import org.apache.hadoop.hdds.utils.db.cache.TableCacheUpdateTracker;
import org.apache.hadoop.ozone.OzoneAcl;
import org.apache.hadoop.ozone.om.OzoneAclUtils;
import org.apache.hadoop.ozone.om.OzoneManager;
@@ -459,11 +460,12 @@ public OMClientResponse handleWriteRequestImpl(OMRequest
omRequest, ExecutionCon
injectPause();
OMClientRequest omClientRequest =
OzoneManagerRatisUtils.createClientRequest(omRequest, impl);
- try {
+ try (TableCacheUpdateTracker tracker = TableCacheUpdateTracker.track()) {
OMClientResponse omClientResponse = captureLatencyNs(
impl.getPerfMetrics().getValidateAndUpdateCacheLatencyNs(),
() ->
Objects.requireNonNull(omClientRequest.validateAndUpdateCache(getOzoneManager(),
context),
"omClientResponse returned by validateAndUpdateCache cannot be
null"));
+ omClientResponse.addCleanupTables(tracker.removeUpdatedTables());
OMAuditLogger.log(omClientRequest.getAuditBuilder(),
context.getTermIndex());
return omClientResponse;
} catch (Throwable th) {
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBufferWithDummyResponse.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBufferWithDummyResponse.java
index 21af533b545..34eb8a7a944 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBufferWithDummyResponse.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBufferWithDummyResponse.java
@@ -20,18 +20,23 @@
import static org.apache.hadoop.hdds.HddsConfigKeys.OZONE_METADATA_DIRS;
import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.BUCKET_TABLE;
+import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.VOLUME_TABLE;
import static org.apache.ozone.test.GenericTestUtils.waitFor;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import com.google.common.collect.Iterators;
import java.io.IOException;
import java.nio.file.Path;
+import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
+import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
@@ -96,22 +101,32 @@ public void testDoubleBufferWithDummyResponse() throws
Exception {
assertEquals(0, metrics.getTotalNumOfFlushedTransactions());
assertEquals(0, metrics.getMaxNumberOfTransactionsFlushedInOneIteration());
+ OMDummyCreateBucketResponse annotatedResponse =
+ createDummyBucketResponse(UUID.randomUUID().toString());
+ annotatedResponse.addCleanupTables(Collections.singleton(VOLUME_TABLE));
+ assertAnnotatedCleanupTakesPrecedence(annotatedResponse);
+
+ OMTrackedDummyCreateBucketResponse trackedResponse =
+ createTrackedDummyBucketResponse(UUID.randomUUID().toString());
+ assertBucketCacheCleanup(trackedResponse);
+
for (int i = 0; i < bucketCount; i++) {
doubleBuffer.add(createDummyBucketResponse(volumeName),
TermIndex.valueOf(term, trxId.incrementAndGet()));
}
- waitFor(() -> metrics.getTotalNumOfFlushedTransactions() == bucketCount,
+ int totalBucketCount = bucketCount + 2;
+ waitFor(() -> metrics.getTotalNumOfFlushedTransactions() ==
totalBucketCount,
100, 60000);
assertThat(metrics.getTotalNumOfFlushOperations()).isGreaterThan(0);
- assertEquals(bucketCount,
doubleBuffer.getFlushedTransactionCountForTesting());
+ assertEquals(totalBucketCount,
doubleBuffer.getFlushedTransactionCountForTesting());
assertThat(metrics.getMaxNumberOfTransactionsFlushedInOneIteration()).isGreaterThan(0);
- assertEquals(bucketCount, omMetadataManager.countRowsInTable(
+ assertEquals(totalBucketCount, omMetadataManager.countRowsInTable(
omMetadataManager.getBucketTable()));
assertThat(doubleBuffer.getFlushIterationsForTesting()).isGreaterThan(0);
assertThat(metrics.getFlushTime().lastStat().numSamples()).isGreaterThan(0);
assertThat(metrics.getAvgFlushTransactionsInOneIteration()).isGreaterThan(0);
- assertEquals(bucketCount, (long)
metrics.getQueueSize().lastStat().total());
+ assertEquals(totalBucketCount, (long)
metrics.getQueueSize().lastStat().total());
assertThat(metrics.getQueueSize().lastStat().numSamples()).isGreaterThan(0);
// Assert there is only instance of OM Double Metrics.
@@ -123,10 +138,36 @@ public void testDoubleBufferWithDummyResponse() throws
Exception {
omMetadataManager.getTransactionInfoTable().get(TRANSACTION_INFO_KEY);
// Check lastAppliedIndex is updated correctly or not.
assertNotNull(transactionInfo);
- assertEquals(bucketCount, transactionInfo.getTransactionIndex());
+ assertEquals(totalBucketCount, transactionInfo.getTransactionIndex());
assertEquals(term, transactionInfo.getTerm());
}
+ private void assertBucketCacheCleanup(
+ AbstractDummyCreateBucketResponse response) throws Exception {
+ long transactionIndex = trxId.incrementAndGet();
+ OmBucketInfo bucketInfo = response.getOmBucketInfo();
+ String dbBucketKey = omMetadataManager.getBucketKey(
+ bucketInfo.getVolumeName(), bucketInfo.getBucketName());
+ omMetadataManager.getBucketTable().addCacheEntry(
+ new CacheKey<>(dbBucketKey), CacheValue.get(transactionIndex));
+
+ doubleBuffer.add(response, TermIndex.valueOf(term, transactionIndex));
+ waitFor(() ->
Iterators.size(omMetadataManager.getBucketTable().cacheIterator()) == 0, 100,
5000);
+ }
+
+ private void assertAnnotatedCleanupTakesPrecedence(
+ AbstractDummyCreateBucketResponse response) throws Exception {
+ long transactionIndex = trxId.get() + 1;
+ String volumeName = response.getOmBucketInfo().getVolumeName();
+ omMetadataManager.getVolumeTable().addCacheEntry(
+ new CacheKey<>(volumeName), CacheValue.get(transactionIndex));
+
+ assertBucketCacheCleanup(response);
+
+ assertEquals(1,
Iterators.size(omMetadataManager.getVolumeTable().cacheIterator()));
+
omMetadataManager.getVolumeTable().cleanupCache(Collections.singletonList(transactionIndex));
+ }
+
/**
* Create DummyBucketCreate response.
*/
@@ -146,20 +187,38 @@ private OMDummyCreateBucketResponse
createDummyBucketResponse(
.build());
}
+ private OMTrackedDummyCreateBucketResponse createTrackedDummyBucketResponse(
+ String volumeName) {
+ OmBucketInfo omBucketInfo =
+ OmBucketInfo.newBuilder()
+ .setVolumeName(volumeName)
+ .setBucketName(UUID.randomUUID().toString())
+ .setCreationTime(Time.now())
+ .build();
+ return new OMTrackedDummyCreateBucketResponse(omBucketInfo,
+ OMResponse.newBuilder()
+ .setCmdType(OzoneManagerProtocolProtos.Type.CreateBucket)
+ .setStatus(OzoneManagerProtocolProtos.Status.OK)
+ .setCreateBucketResponse(CreateBucketResponse.newBuilder().build())
+ .build());
+ }
/**
* DummyCreatedBucket Response class used in testing.
*/
- @CleanupTableInfo(cleanupTables = {BUCKET_TABLE})
- private static class OMDummyCreateBucketResponse extends OMClientResponse {
+ private abstract static class AbstractDummyCreateBucketResponse extends
OMClientResponse {
private final OmBucketInfo omBucketInfo;
- OMDummyCreateBucketResponse(OmBucketInfo omBucketInfo,
+ AbstractDummyCreateBucketResponse(OmBucketInfo omBucketInfo,
OMResponse omResponse) {
super(omResponse);
this.omBucketInfo = omBucketInfo;
}
+ OmBucketInfo getOmBucketInfo() {
+ return omBucketInfo;
+ }
+
@Override
public void addToDBBatch(OMMetadataManager omMetadataManager,
BatchOperation batchOperation) throws IOException {
@@ -170,4 +229,22 @@ public void addToDBBatch(OMMetadataManager
omMetadataManager,
dbBucketKey, omBucketInfo);
}
}
+
+ @CleanupTableInfo(cleanupTables = {BUCKET_TABLE})
+ private static class OMDummyCreateBucketResponse
+ extends AbstractDummyCreateBucketResponse {
+ OMDummyCreateBucketResponse(OmBucketInfo omBucketInfo,
+ OMResponse omResponse) {
+ super(omBucketInfo, omResponse);
+ }
+ }
+
+ private static class OMTrackedDummyCreateBucketResponse
+ extends AbstractDummyCreateBucketResponse {
+ OMTrackedDummyCreateBucketResponse(OmBucketInfo omBucketInfo,
+ OMResponse omResponse) {
+ super(omBucketInfo, omResponse);
+ addCleanupTables(Collections.singleton(BUCKET_TABLE));
+ }
+ }
}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMOpenKeysDeleteRequest.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMOpenKeysDeleteRequest.java
index 03f9b774411..fef6583ed49 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMOpenKeysDeleteRequest.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMOpenKeysDeleteRequest.java
@@ -17,6 +17,8 @@
package org.apache.hadoop.ozone.om.request.key;
+import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_FILE_TABLE;
+import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_KEY_TABLE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -35,6 +37,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -256,6 +259,10 @@ public void testDeleteKeyWithHigherUpdateID(
openKeyDeleteRequest.validateAndUpdateCache(ozoneManager,
transactionId);
assertEquals(Status.OK, omClientResponse.getOMResponse().getStatus());
+ String expectedCleanupTable =
+ getBucketLayout().isFileSystemOptimized() ? OPEN_FILE_TABLE :
OPEN_KEY_TABLE;
+ assertEquals(Collections.singleton(expectedCleanupTable),
+ omClientResponse.removeCleanupTables());
assertInOpenKeyTable(keysWithHigherUpdateID);
assertNotInOpenKeyTable(keysWithSameUpdateID);
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java
index 5de300e2c2b..fb8b2eb6f77 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestCleanupTableInfo.java
@@ -62,6 +62,7 @@
import org.apache.hadoop.ozone.om.request.key.OMKeyCreateRequest;
import org.apache.hadoop.ozone.om.response.file.OMFileCreateResponse;
import org.apache.hadoop.ozone.om.response.key.OMKeyCreateResponse;
+import org.apache.hadoop.ozone.om.response.key.OMOpenKeysDeleteResponse;
import
org.apache.hadoop.ozone.om.response.lifecycle.OMLifecycleSetServiceStatusResponse;
import org.apache.hadoop.ozone.om.response.s3.security.S3AssumeRoleResponse;
import org.apache.hadoop.ozone.om.response.util.OMEchoRPCWriteResponse;
@@ -141,7 +142,9 @@ public void checkAnnotationAndTableName() {
subTypes.remove(OMEchoRPCWriteResponse.class);
subTypes.remove(DummyOMClientResponse.class);
subTypes.remove(OMLifecycleSetServiceStatusResponse.class);
+ subTypes.remove(OMOpenKeysDeleteResponse.class);
subTypes.remove(S3AssumeRoleResponse.class);
+
assertFalse(OMOpenKeysDeleteResponse.class.isAnnotationPresent(CleanupTableInfo.class));
subTypes.forEach(aClass -> {
if (Modifier.isAbstract(aClass.getModifiers())) {
assertFalse(aClass.isAnnotationPresent(CleanupTableInfo.class),
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestOMClientResponse.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestOMClientResponse.java
new file mode 100644
index 00000000000..3c68407682c
--- /dev/null
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/response/TestOMClientResponse.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.response;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests generic OM client response behavior.
+ */
+public class TestOMClientResponse {
+
+ @Test
+ public void testCleanupTablesMergeTrackedAndResponseTables() {
+ OMClientResponse response = new DummyOMClientResponse(
+ OMResponse.newBuilder()
+ .setStatus(Status.OK)
+ .setCmdType(Type.CreateSnapshot)
+ .build());
+
+ response.addCleanupTables(new LinkedHashSet<>(Arrays.asList(
+ "deletedTable", "snapshotRenamedTable")));
+ response.addCleanupTables(new LinkedHashSet<>(
+ Arrays.asList("snapshotInfoTable")));
+
+ assertEquals(new LinkedHashSet<>(Arrays.asList("deletedTable",
+ "snapshotRenamedTable", "snapshotInfoTable")),
+ response.removeCleanupTables());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
