divide FSNameSystem.addSafeBlocks in several requests
import static io.hops.transaction.lock.LockFactory.BLK;
import static io.hops.transaction.lock.LockFactory.getInstance;
import io.hops.transaction.lock.TransactionLockTypes;
+import io.hops.util.Slicer;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.impl.Log4JLogger;
import org.apache.hadoop.fs.BatchedRemoteIterator.BatchedListEntries;
import org.apache.hadoop.fs.CacheFlag;
@@ -478,6 +480,8 @@ private void logAuditEvent(boolean succeeded, UserGroupInformation ugi,
long delayBeforeSTOFlag = 0; //This parameter can not be more than TxInactiveTimeout: 1.2 sec
long delayAfterBuildingTree=0;
+ int slicerBatchSize;
+ int slicerNbThreads;
/**
* Clear all loaded data
*/
@@ -675,6 +679,12 @@ private FSNamesystem(Configuration conf, NameNode namenode, boolean ignoreRetryC
auditLoggers.get(0) instanceof DefaultAuditLogger;
this.aclConfigFlag = new AclConfigFlag(conf);
this.retryCache = ignoreRetryCache ? null : initRetryCache(conf);
+ this.slicerBatchSize = conf.getInt(DFSConfigKeys.DFS_NAMENODE_PROCESS_MISREPLICATED_BATCH_SIZE,
+ DFSConfigKeys.DFS_NAMENODE_PROCESS_MISREPLICATED_BATCH_SIZE_DEFAULT);
+
+ this.slicerNbThreads = conf.getInt(
+ DFSConfigKeys.DFS_NAMENODE_PROCESS_MISREPLICATED_NO_OF_THREADS,
+ DFSConfigKeys.DFS_NAMENODE_PROCESS_MISREPLICATED_NO_OF_THREADS_DEFAULT);
} catch (IOException | RuntimeException e) {
LOG.error(getClass().getSimpleName() + " initialization failed.", e);
close();
@@ -5271,7 +5281,7 @@ private void setSafeModePendingOperation(Boolean val) {
private void adjustSafeBlocks(Set<Long> safeBlocks) throws IOException {
int lastSafeBlockSize = blockSafe();
- addSafeBlocks(safeBlocks);
+ addSafeBlocks(new ArrayList<Long>(safeBlocks));
int newSafeBlockSize = blockSafe();
if (LOG.isDebugEnabled()) {
long blockTotal = blockTotal();
@@ -7547,7 +7557,7 @@ QuotaUpdateManager getQuotaUpdateManager() {
* @throws IOException
*/
private void addSafeBlock(final Long safeBlock) throws IOException {
- Set<Long> safeBlocks = new HashSet<>();
+ List<Long> safeBlocks = new ArrayList();
safeBlocks.add(safeBlock);
addSafeBlocks(safeBlocks);
}
@@ -7576,17 +7586,38 @@ public Object performTask() throws IOException {
* list of blocks to be added to safe blocks
* @throws IOException
*/
- private void addSafeBlocks(final Set<Long> safeBlocks) throws IOException {
+ private void addSafeBlocks(final List<Long> safeBlocks) throws IOException {
+ try {
+ Slicer.slice(safeBlocks.size(), slicerBatchSize, slicerNbThreads,
+ new Slicer.OperationHandler() {
+ @Override
+ public void handle(int startIndex, int endIndex) throws Exception {
+ final List<Long> ids = safeBlocks.subList(startIndex, endIndex);
new LightWeightRequestHandler(HDFSOperationType.ADD_SAFE_BLOCKS) {
@Override
public Object performTask() throws IOException {
+ boolean inTransaction = connector.isTransactionActive();
+ if (!inTransaction) {
+ connector.beginTransaction();
+ }
SafeBlocksDataAccess da = (SafeBlocksDataAccess) HdfsStorageFactory
.getDataAccess(SafeBlocksDataAccess.class);
- da.insert(safeBlocks);
+ da.insert(ids);
+ if(!inTransaction){
+ connector.commit();
+ }
return null;
}
}.handle();
}
+ });
+ } catch (Exception e) {
+ if (e instanceof IOException) {
+ throw (IOException) e;
+ }
+ throw new IOException(e);
+ }
+ }
/**
* Get number of blocks to be considered safe in the current cluster
