replicatedSizeMap;
+ private final int maxKeysInMemory;
+ private final int maxIterators;
@Inject
public OmTableInsightTask(ReconGlobalStatsManager reconGlobalStatsManager,
@@ -72,6 +80,12 @@ public OmTableInsightTask(ReconGlobalStatsManager reconGlobalStatsManager,
tableHandlers.put(OPEN_FILE_TABLE, new OpenKeysInsightHandler());
tableHandlers.put(DELETED_TABLE, new DeletedKeysInsightHandler());
tableHandlers.put(MULTIPART_INFO_TABLE, new MultipartInfoInsightHandler());
+ this.maxKeysInMemory = reconOMMetadataManager.getOzoneConfiguration().getInt(
+ ReconServerConfigKeys.OZONE_RECON_TASK_REPROCESS_MAX_KEYS_IN_MEMORY,
+ ReconServerConfigKeys.OZONE_RECON_TASK_REPROCESS_MAX_KEYS_IN_MEMORY_DEFAULT);
+ this.maxIterators = reconOMMetadataManager.getOzoneConfiguration().getInt(
+ ReconServerConfigKeys.OZONE_RECON_TASK_REPROCESS_MAX_ITERATORS,
+ ReconServerConfigKeys.OZONE_RECON_TASK_REPROCESS_MAX_ITERATORS_DEFAULT);
}
@Override
@@ -97,39 +111,36 @@ public void init() {
}
/**
- * Iterates the rows of each table in the OM snapshot DB and calculates the
- * counts and sizes for table data.
- *
- * For tables that require data size calculation
- * (as returned by getTablesToCalculateSize), both the number of
- * records (count) and total data size of the records are calculated.
- * For all other tables, only the count of records is calculated.
+ * Reprocess all OM tables to calculate counts and sizes.
+ * Handler tables (with size calculation) use sequential iteration.
+ * Simple tables (count only) use parallel iteration with String keys,
+ * or sequential for non-String key tables.
*
- * @param omMetadataManager OM Metadata instance.
- * @return Pair
+ * @param omMetadataManager OM Metadata instance
+ * @return TaskResult indicating success or failure
*/
@Override
public TaskResult reprocess(OMMetadataManager omMetadataManager) {
+ LOG.info("{}: Starting reprocess", getTaskName());
+ long startTime = Time.monotonicNow();
+
init();
for (String tableName : tables) {
- Table table = omMetadataManager.getTable(tableName);
-
- try (TableIterator> iterator
- = table.iterator()) {
+ try {
if (tableHandlers.containsKey(tableName)) {
- Triple details =
- tableHandlers.get(tableName).getTableSizeAndCount(iterator);
- objectCountMap.put(getTableCountKeyFromTable(tableName),
- details.getLeft());
- unReplicatedSizeMap.put(
- getUnReplicatedSizeKeyFromTable(tableName), details.getMiddle());
- replicatedSizeMap.put(getReplicatedSizeKeyFromTable(tableName),
- details.getRight());
+ Triple details =
+ tableHandlers.get(tableName).getTableSizeAndCount(tableName, omMetadataManager);
+ objectCountMap.put(getTableCountKeyFromTable(tableName), details.getLeft());
+ unReplicatedSizeMap.put(getUnReplicatedSizeKeyFromTable(tableName), details.getMiddle());
+ replicatedSizeMap.put(getReplicatedSizeKeyFromTable(tableName), details.getRight());
} else {
- long count = Iterators.size(iterator);
- objectCountMap.put(getTableCountKeyFromTable(tableName), count);
+ if (usesNonStringKeys(tableName)) {
+ processTableSequentially(tableName, omMetadataManager);
+ } else {
+ processTableInParallel(tableName, omMetadataManager);
+ }
}
- } catch (IOException ioEx) {
+ } catch (Exception ioEx) {
LOG.error("Unable to populate Table Count in Recon DB.", ioEx);
return buildTaskResult(false);
}
@@ -144,11 +155,79 @@ public TaskResult reprocess(OMMetadataManager omMetadataManager) {
if (!replicatedSizeMap.isEmpty()) {
writeDataToDB(replicatedSizeMap);
}
+ long endTime = Time.monotonicNow();
+ long durationMs = endTime - startTime;
- LOG.debug("Completed a 'reprocess' run of OmTableInsightTask.");
+ LOG.info("{}: Reprocess completed in {} ms", getTaskName(), durationMs);
return buildTaskResult(true);
}
+ /**
+ * Check if table uses non-String keys (e.g., OzoneTokenIdentifier).
+ * These tables cannot use StringCodec and must be processed sequentially.
+ */
+ private boolean usesNonStringKeys(String tableName) {
+ return tableName.equals("dTokenTable") || tableName.equals("s3SecretTable");
+ }
+
+ /**
+ * Process table sequentially using key-only iterator.
+ * Used for tables with non-String keys or as fallback.
+ */
+ private void processTableSequentially(String tableName, OMMetadataManager omMetadataManager) throws IOException {
+ LOG.info("{}: Processing table {} sequentially (non-String keys)", getTaskName(), tableName);
+
+ Table table = omMetadataManager.getStore()
+ .getTable(tableName, ByteArrayCodec.get(), ByteArrayCodec.get(), TableCache.CacheType.NO_CACHE);
+ try (TableIterator keyIterator = table.keyIterator()) {
+ long count = Iterators.size(keyIterator);
+ objectCountMap.put(getTableCountKeyFromTable(tableName), count);
+ }
+ }
+
+ /**
+ * Process table in parallel using multiple iterators and workers.
+ * Only for tables with String keys.
+ */
+ private void processTableInParallel(String tableName, OMMetadataManager omMetadataManager) throws Exception {
+ int workerCount = 2; // Only 2 workers needed for simple counting
+
+ Table table = omMetadataManager.getStore()
+ .getTable(tableName, StringCodec.get(), ByteArrayCodec.get(), TableCache.CacheType.NO_CACHE);
+
+ long estimatedCount = 100000; // Default
+ try {
+ estimatedCount = table.getEstimatedKeyCount();
+ } catch (IOException e) {
+ LOG.info("Could not estimate key count for table {}, using default", tableName);
+ }
+ long loggingThreshold = calculateLoggingThreshold(estimatedCount);
+
+ AtomicLong count = new AtomicLong(0);
+
+ try (ParallelTableIteratorOperation parallelIter = new ParallelTableIteratorOperation<>(
+ omMetadataManager, table, StringCodec.get(),
+ maxIterators, workerCount, maxKeysInMemory, loggingThreshold)) {
+
+ parallelIter.performTaskOnTableVals(getTaskName(), null, null, kv -> {
+ if (kv != null) {
+ count.incrementAndGet();
+ }
+ return null;
+ });
+ }
+
+ objectCountMap.put(getTableCountKeyFromTable(tableName), count.get());
+ }
+
+ /**
+ * Calculate logging threshold based on estimated key count.
+ * Logs progress every 1% of total keys, minimum 1.
+ */
+ private long calculateLoggingThreshold(long estimatedCount) {
+ return Math.max(estimatedCount / 100, 1);
+ }
+
@Override
public String getTaskName() {
return "OmTableInsightTask";
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OpenKeysInsightHandler.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OpenKeysInsightHandler.java
index d22963ed2807..b78e8cb1518f 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OpenKeysInsightHandler.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OpenKeysInsightHandler.java
@@ -22,6 +22,7 @@
import org.apache.commons.lang3.tuple.Triple;
import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -127,18 +128,18 @@ public void handleUpdateEvent(OMDBUpdateEvent event,
* that are currently open in the backend.
*/
@Override
- public Triple getTableSizeAndCount(
- TableIterator> iterator)
- throws IOException {
+ public Triple getTableSizeAndCount(String tableName,
+ OMMetadataManager omMetadataManager) throws IOException {
long count = 0;
long unReplicatedSize = 0;
long replicatedSize = 0;
- if (iterator != null) {
+ Table table = (Table) omMetadataManager.getTable(tableName);
+ try (TableIterator> iterator = table.iterator()) {
while (iterator.hasNext()) {
- Table.KeyValue kv = iterator.next();
+ Table.KeyValue kv = iterator.next();
if (kv != null && kv.getValue() != null) {
- OmKeyInfo omKeyInfo = (OmKeyInfo) kv.getValue();
+ OmKeyInfo omKeyInfo = kv.getValue();
unReplicatedSize += omKeyInfo.getDataSize();
replicatedSize += omKeyInfo.getReplicatedSize();
count++;
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java
index 69f866239878..971614325829 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/ReconTaskControllerImpl.java
@@ -218,6 +218,7 @@ public synchronized boolean reInitializeTasks(ReconOMMetadataManager omMetadataM
});
AtomicBoolean isRunSuccessful = new AtomicBoolean(true);
+ LOG.info("Submitting {} tasks for parallel reprocessing", tasks.size());
try {
CompletableFuture.allOf(tasks.stream()
.map(task -> {
@@ -225,8 +226,11 @@ public synchronized boolean reInitializeTasks(ReconOMMetadataManager omMetadataM
long reprocessStartTime = Time.monotonicNow();
return CompletableFuture.supplyAsync(() -> {
+ LOG.info("Task {} started execution on thread {}",
+ task.getTaskName(), Thread.currentThread().getName());
try {
ReconOmTask.TaskResult result = task.call();
+ LOG.info("Task {} completed execution", task.getTaskName());
return result;
} catch (Exception e) {
// Track reprocess failure per task
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/ParallelTableIteratorOperation.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/ParallelTableIteratorOperation.java
new file mode 100644
index 000000000000..c91a3d97775b
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/ParallelTableIteratorOperation.java
@@ -0,0 +1,296 @@
+/*
+ * 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.recon.tasks.util;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.hdds.utils.db.RDBStore;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.rocksdb.LiveFileMetaData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Class to iterate through a table in parallel by breaking table into multiple iterators.
+ */
+public class ParallelTableIteratorOperation, V> implements Closeable {
+ private final Table table;
+ private final Codec keyCodec;
+
+ // Thread Pools
+ private final ExecutorService iteratorExecutor; // 5
+ private final ExecutorService valueExecutors; // 20
+
+ private final int maxNumberOfVals;
+ private final OMMetadataManager metadataManager;
+ private final int maxIteratorTasks;
+ private final int maxWorkerTasks;
+ private final long logCountThreshold;
+
+ private static final Logger LOG = LoggerFactory.getLogger(ParallelTableIteratorOperation.class);
+
+ public ParallelTableIteratorOperation(OMMetadataManager metadataManager, Table table, Codec keyCodec,
+ int iteratorCount, int workerCount, int maxNumberOfValsInMemory,
+ long logThreshold) {
+ this.table = table;
+ this.keyCodec = keyCodec;
+ this.metadataManager = metadataManager;
+ this.maxIteratorTasks = 2 * iteratorCount; // Allow up to 10 pending iterator tasks
+ this.maxWorkerTasks = workerCount * 2; // Allow up to 40 pending worker tasks
+
+ // Create team of 5 iterator threads with UNLIMITED queue
+ // LinkedBlockingQueue() with no size = can hold infinite pending tasks
+ this.iteratorExecutor = new ThreadPoolExecutor(iteratorCount, iteratorCount, 1, TimeUnit.MINUTES,
+ new LinkedBlockingQueue<>());
+
+ // Create team of 20 worker threads with UNLIMITED queue
+ this.valueExecutors = new ThreadPoolExecutor(workerCount, workerCount, 1, TimeUnit.MINUTES,
+ new LinkedBlockingQueue<>());
+
+ // Calculate batch size per worker (e.g., 2000 / 20 = 100 keys per batch per worker)
+ this.maxNumberOfVals = Math.max(10, maxNumberOfValsInMemory / (workerCount));
+ this.logCountThreshold = logThreshold;
+ }
+
+ private List getBounds(K startKey, K endKey) throws IOException {
+ Set keys = new HashSet<>();
+
+ // Try to get SST file boundaries for optimal segmentation
+ // In test/mock environments, this may not be available
+ try {
+ RDBStore store = (RDBStore) this.metadataManager.getStore();
+ if (store != null && store.getDb() != null) {
+ List sstFiles = store.getDb().getSstFileList();
+ String tableName = table.getName();
+
+ // Only filter by column family if table name is available
+ if (tableName != null && !tableName.isEmpty()) {
+ byte[] tableNameBytes = tableName.getBytes(StandardCharsets.UTF_8);
+ for (LiveFileMetaData sstFile : sstFiles) {
+ // Filter SST files by column family to get bounds only for this specific table
+ if (Arrays.equals(sstFile.columnFamilyName(), tableNameBytes)) {
+ keys.add(this.keyCodec.fromPersistedFormat(sstFile.smallestKey()));
+ keys.add(this.keyCodec.fromPersistedFormat(sstFile.largestKey()));
+ }
+ }
+ }
+ }
+ } catch (Exception e) {
+ // If we can't get SST files (test environment, permissions, etc.),
+ // just use empty bounds and rely on fallback path
+ LOG.debug("Unable to retrieve SST file boundaries, will use fallback iteration: {}", e.getMessage());
+ }
+
+ if (startKey != null) {
+ keys.add(startKey);
+ }
+ if (endKey != null) {
+ keys.add(endKey);
+ }
+
+ return keys.stream().sorted().filter(Objects::nonNull)
+ .filter(key -> startKey == null || key.compareTo(startKey) >= 0)
+ .filter(key -> endKey == null || endKey.compareTo(key) >= 0)
+ .collect(Collectors.toList());
+ }
+
+ private void waitForQueueSize(Queue> futures, int expectedSize)
+ throws ExecutionException, InterruptedException {
+ while (!futures.isEmpty() && futures.size() > expectedSize) {
+ Future f = futures.poll();
+ f.get();
+ }
+ }
+
+ // Main parallelization logic
+ public void performTaskOnTableVals(String taskName, K startKey, K endKey,
+ Function, Void> keyOperation) throws IOException, ExecutionException, InterruptedException {
+ List bounds = getBounds(startKey, endKey);
+
+ // Fallback for small tables (no SST files yet - data only in memtable)
+ if (bounds.size() < 2) {
+ try (TableIterator> iter = table.iterator()) {
+ if (startKey != null) {
+ iter.seek(startKey);
+ }
+ while (iter.hasNext()) {
+ Table.KeyValue kv = iter.next();
+ if (endKey != null && kv.getKey().compareTo(endKey) > 0) {
+ break;
+ }
+ keyOperation.apply(kv);
+ }
+ }
+ return;
+ }
+
+ // ===== PARALLEL PROCESSING SETUP =====
+
+ // Queue to track iterator threads (5 threads creating work)
+ Queue> iterFutures = new LinkedList<>();
+
+ // Queue to track worker threads (20 threads doing work)
+ Queue> workerFutures = new ConcurrentLinkedQueue<>();
+
+ AtomicLong keyCounter = new AtomicLong();
+ AtomicLong prevLogCounter = new AtomicLong();
+ Object logLock = new Object();
+
+ // ===== STEP 2: START ITERATOR THREADS =====
+ // For each segment boundary, create an iterator thread
+ // Example: If bounds = [0, 5M, 10M, 15M, 20M], this loop runs 4 times:
+ // idx=1: beg=0, end=5M
+ // idx=2: beg=5M, end=10M
+ // idx=3: beg=10M, end=15M
+ // idx=4: beg=15M, end=20M
+ for (int idx = 1; idx < bounds.size(); idx++) {
+ K beg = bounds.get(idx - 1);
+ K end = bounds.get(idx);
+ boolean inclusive = idx == bounds.size() - 1;
+ waitForQueueSize(iterFutures, maxIteratorTasks - 1);
+
+ // ===== STEP 3: SUBMIT ITERATOR TASK =====
+ iterFutures.add(iteratorExecutor.submit(() -> {
+ try (TableIterator> iter = table.iterator()) {
+ iter.seek(beg);
+ while (iter.hasNext()) {
+ List> keyValues = new ArrayList<>();
+ boolean reachedEnd = false;
+ while (iter.hasNext()) {
+ Table.KeyValue kv = iter.next();
+ K key = kv.getKey();
+
+ // Check if key is within this segment's range
+ boolean withinBounds;
+ if (inclusive) {
+ // Last segment: include everything from beg onwards (or until endKey if specified)
+ withinBounds = (endKey == null || key.compareTo(endKey) <= 0);
+ } else {
+ // Middle segment: include keys in range [beg, end)
+ withinBounds = key.compareTo(end) < 0;
+ }
+
+ if (withinBounds) {
+ keyValues.add(kv);
+ } else {
+ reachedEnd = true;
+ break;
+ }
+
+ // If batch is full (2000 keys), stop collecting
+ if (keyValues.size() >= maxNumberOfVals) {
+ break;
+ }
+ }
+
+ // ===== STEP 5: HAND BATCH TO WORKER THREAD =====
+ if (!keyValues.isEmpty()) {
+ // WAIT if worker queue is too full (max 39 pending tasks)
+ waitForQueueSize(workerFutures, maxWorkerTasks - 1);
+
+ // Submit batch to worker thread pool
+ workerFutures.add(valueExecutors.submit(() -> {
+ for (Table.KeyValue kv : keyValues) {
+ keyOperation.apply(kv);
+ }
+ keyCounter.addAndGet(keyValues.size());
+ if (keyCounter.get() - prevLogCounter.get() > logCountThreshold) {
+ synchronized (logLock) {
+ if (keyCounter.get() - prevLogCounter.get() > logCountThreshold) {
+ long cnt = keyCounter.get();
+ LOG.debug("Iterated through {} keys while performing task: {}", keyCounter.get(), taskName);
+ prevLogCounter.set(cnt);
+ }
+ }
+ }
+ // Worker task done! Future is now complete.
+ }));
+ }
+ // If we reached the end of our segment, stop reading
+ if (reachedEnd) {
+ break;
+ }
+ }
+ } catch (IOException e) {
+ LOG.error("IO error during parallel iteration on table {}", taskName, e);
+ throw new RuntimeException("IO error during iteration", e);
+ } catch (InterruptedException e) {
+ LOG.warn("Parallel iteration interrupted for task {}", taskName, e);
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Iteration interrupted", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ LOG.error("Task execution failed for {}: {}", taskName, cause.getMessage(), cause);
+ throw new RuntimeException("Task execution failed", cause);
+ }
+ }));
+ }
+
+ // ===== STEP 7: WAIT FOR EVERYONE TO FINISH =====
+ // Wait for all 5 iterator threads to finish reading
+ waitForQueueSize(iterFutures, 0);
+ // Wait for all 20 worker threads to finish processing
+ waitForQueueSize(workerFutures, 0);
+
+ // Log final stats
+ LOG.info("{}: Parallel iteration completed - Total keys processed: {}", taskName, keyCounter.get());
+ }
+
+ @Override
+ public void close() throws IOException {
+ iteratorExecutor.shutdown();
+ valueExecutors.shutdown();
+ try {
+ if (!iteratorExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
+ iteratorExecutor.shutdownNow();
+ }
+ if (!valueExecutors.awaitTermination(60, TimeUnit.SECONDS)) {
+ valueExecutors.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ iteratorExecutor.shutdownNow();
+ valueExecutors.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+}
+
+
+
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/package-info.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/package-info.java
new file mode 100644
index 000000000000..f8ec57de2f2f
--- /dev/null
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/util/package-info.java
@@ -0,0 +1,26 @@
+/*
+ * 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 to define utility classes for tasks.
+ */
+package org.apache.hadoop.ozone.recon.tasks.util;
+
+
+
+
+
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java
index 4794ecf1f309..7df56a57be65 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java
@@ -47,6 +47,10 @@
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import javax.ws.rs.WebApplicationException;
@@ -77,6 +81,7 @@
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.recon.ReconConstants;
import org.apache.hadoop.ozone.recon.ReconTestInjector;
import org.apache.hadoop.ozone.recon.api.types.ContainerDiscrepancyInfo;
import org.apache.hadoop.ozone.recon.api.types.ContainerKeyPrefix;
@@ -100,9 +105,11 @@
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
import org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl;
import org.apache.hadoop.ozone.recon.spi.impl.StorageContainerServiceProviderImpl;
+import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperHelper;
import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperTaskFSO;
import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperTaskOBS;
import org.apache.hadoop.ozone.recon.tasks.NSSummaryTaskWithFSO;
+import org.apache.hadoop.ozone.recon.tasks.ReconOmTask;
import org.apache.ozone.recon.schema.ContainerSchemaDefinition.UnHealthyContainerStates;
import org.apache.ozone.recon.schema.generated.tables.pojos.UnhealthyContainers;
import org.junit.jupiter.api.BeforeEach;
@@ -216,7 +223,15 @@ public void setUp() throws Exception {
if (!isSetupDone) {
initializeInjector();
isSetupDone = true;
+ } else {
+ // Clear shared state before subsequent tests to prevent data leakage
+ ContainerKeyMapperHelper.clearSharedContainerCountMap();
+ ReconConstants.resetTableTruncatedFlags();
+
+ // Reinitialize container tables to clear RocksDB data
+ reconContainerMetadataManager.reinitWithNewContainerDataFromOm(Collections.emptyMap());
}
+
omConfiguration = new OzoneConfiguration();
List omKeyLocationInfoList = new ArrayList<>();
@@ -297,14 +312,27 @@ public void setUp() throws Exception {
reprocessContainerKeyMapper();
}
- private void reprocessContainerKeyMapper() {
+ private void reprocessContainerKeyMapper() throws Exception {
ContainerKeyMapperTaskOBS containerKeyMapperTaskOBS =
new ContainerKeyMapperTaskOBS(reconContainerMetadataManager, omConfiguration);
- containerKeyMapperTaskOBS.reprocess(reconOMMetadataManager);
-
ContainerKeyMapperTaskFSO containerKeyMapperTaskFSO =
new ContainerKeyMapperTaskFSO(reconContainerMetadataManager, omConfiguration);
- containerKeyMapperTaskFSO.reprocess(reconOMMetadataManager);
+
+ // Run both tasks in parallel (like production)
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ Future obsFuture = executor.submit(
+ () -> containerKeyMapperTaskOBS.reprocess(reconOMMetadataManager));
+ Future fsoFuture = executor.submit(
+ () -> containerKeyMapperTaskFSO.reprocess(reconOMMetadataManager));
+
+ // Wait for both to complete
+ obsFuture.get();
+ fsoFuture.get();
+ } finally {
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ }
}
private void setUpFSOData() throws IOException {
@@ -435,7 +463,7 @@ private OmKeyLocationInfoGroup getLocationInfoGroup1() {
}
@Test
- public void testGetKeysForContainer() throws IOException {
+ public void testGetKeysForContainer() throws Exception {
Response response = containerEndpoint.getKeysForContainer(1L, -1, "");
KeysResponse data = (KeysResponse) response.getEntity();
@@ -513,7 +541,7 @@ public void testGetKeysForContainer() throws IOException {
}
@Test
- public void testGetKeysForContainerWithPrevKey() throws IOException {
+ public void testGetKeysForContainerWithPrevKey() throws Exception {
// test if prev-key param works as expected
Response response = containerEndpoint.getKeysForContainer(
1L, -1, "/sampleVol/bucketOne/key_one");
@@ -1370,7 +1398,7 @@ public void testGetContainerInsightsNonSCMContainers()
@Test
public void testGetContainerInsightsNonSCMContainersWithPrevKey()
- throws IOException, TimeoutException {
+ throws Exception {
// Add 3 more containers to OM making total container in OM to 5
String[] keys = {"key_three", "key_four", "key_five"};
@@ -1821,7 +1849,7 @@ private void setUpDuplicateFSOFileKeys() throws IOException {
* and then verifies that the ContainerEndpoint returns two distinct key records.
*/
@Test
- public void testDuplicateFSOKeysForContainerEndpoint() throws IOException {
+ public void testDuplicateFSOKeysForContainerEndpoint() throws Exception {
// Set up duplicate FSO file keys.
setUpDuplicateFSOFileKeys();
NSSummaryTaskWithFSO nSSummaryTaskWithFso =
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java
index 653de02cdbc2..ce8ab5ce2162 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestContainerKeyMapperTask.java
@@ -44,6 +44,7 @@
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
+import org.apache.hadoop.ozone.recon.ReconConstants;
import org.apache.hadoop.ozone.recon.ReconTestInjector;
import org.apache.hadoop.ozone.recon.api.types.ContainerKeyPrefix;
import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
@@ -95,6 +96,10 @@ public void setUp() throws Exception {
.build();
reconContainerMetadataManager =
reconTestInjector.getInstance(ReconContainerMetadataManager.class);
+
+ // Clear shared container count map and reset flags for clean test state
+ ContainerKeyMapperHelper.clearSharedContainerCountMap();
+ ReconConstants.resetTableTruncatedFlags();
}
@Test
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java
index ca4a0399be80..c4b1041f76af 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestFileSizeCountTask.java
@@ -134,6 +134,7 @@ public void testReprocess() throws IOException {
// Note: Even though legacy and OBS share the same underlying table, we simulate OBS here.
when(omMetadataManager.getKeyTable(eq(BucketLayout.OBJECT_STORE)))
.thenReturn(keyTableOBS);
+ when(keyTableOBS.getName()).thenReturn("keyTable"); // Mock table name for parallelization
TypedTable.TypedTableIterator mockIterOBS = mock(TypedTable.TypedTableIterator.class);
when(keyTableOBS.iterator()).thenReturn(mockIterOBS);
// Simulate three keys then end.
@@ -146,6 +147,7 @@ public void testReprocess() throws IOException {
TypedTable keyTableFSO = mock(TypedTable.class);
when(omMetadataManager.getKeyTable(eq(BucketLayout.FILE_SYSTEM_OPTIMIZED)))
.thenReturn(keyTableFSO);
+ when(keyTableFSO.getName()).thenReturn("fileTable"); // Mock table name for parallelization
TypedTable.TypedTableIterator mockIterFSO = mock(TypedTable.TypedTableIterator.class);
when(keyTableFSO.iterator()).thenReturn(mockIterFSO);
when(mockIterFSO.hasNext()).thenReturn(true, true, true, false);
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
index faa158bfab3d..4f64d27297b2 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
@@ -38,6 +38,8 @@
import static org.apache.ozone.recon.schema.generated.tables.GlobalStatsTable.GLOBAL_STATS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -53,6 +55,7 @@
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.utils.db.DBStore;
import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.hdds.utils.db.TypedTable;
import org.apache.hadoop.ozone.om.OMMetadataManager;
@@ -346,6 +349,25 @@ public void testProcessForDeletedDirectoryTable() throws IOException {
@Test
public void testReprocessForCount() throws Exception {
OMMetadataManager omMetadataManager = mock(OmMetadataManagerImpl.class);
+
+ // Mock DBStore for getStore() calls
+ DBStore mockStore = mock(DBStore.class);
+ when(omMetadataManager.getStore()).thenReturn(mockStore);
+
+ // Mock getDeletedTable() for DeletedKeysInsightHandler
+ TypedTable deletedTable = mock(TypedTable.class);
+ TypedTable.TypedTableIterator deletedIter = mock(TypedTable.TypedTableIterator.class);
+ when(deletedTable.iterator()).thenReturn(deletedIter);
+ when(omMetadataManager.getDeletedTable()).thenReturn(deletedTable);
+ when(deletedIter.hasNext()).thenReturn(true, true, true, true, true, false);
+
+ RepeatedOmKeyInfo deletedKeyInfo = mock(RepeatedOmKeyInfo.class);
+ when(deletedKeyInfo.getTotalSize()).thenReturn(ImmutablePair.of(100L, 100L));
+ when(deletedKeyInfo.getOmKeyInfoList()).thenReturn(Arrays.asList(mock(OmKeyInfo.class)));
+
+ Table.KeyValue deletedKv = mock(Table.KeyValue.class);
+ when(deletedKv.getValue()).thenReturn(deletedKeyInfo);
+ when(deletedIter.next()).thenReturn(deletedKv);
// Mock 5 rows in each table and test the count
for (String tableName : omTableInsightTask.getTaskTables()) {
@@ -353,8 +375,13 @@ public void testReprocessForCount() throws Exception {
TypedTable.TypedTableIterator mockIter =
mock(TypedTable.TypedTableIterator.class);
when(table.iterator()).thenReturn(mockIter);
+ when(table.keyIterator()).thenReturn(mockIter);
+ when(table.getEstimatedKeyCount()).thenReturn(5L);
when(omMetadataManager.getTable(tableName)).thenReturn(table);
when(mockIter.hasNext()).thenReturn(true, true, true, true, true, false);
+
+ // Mock DBStore.getTable() to return the same table
+ when(mockStore.getTable(eq(tableName), any(), any(), any())).thenAnswer(invocation -> table);
final Table.KeyValue mockKeyValue = mock(Table.KeyValue.class);