HOPSFS-162: Fixed Retry Cache Cleaner (hopshadoop#64)

* HOPSFS-162: Fixed Retry Cache Cleaner

* HOPSFS-162: create an index on hdfs_retry_cache_entry.epoch

Author: smkniazi

schema/update-schema_3.2.0.12_to_3.2.0.13.sql

@@ -0,0 +1,9 @@
+-- create an index on hdfs_retry_cache_entry.epoch
+-- check if the create index fix is not already applied 
+
+SELECT COUNT(*) INTO @index_exists FROM information_schema.statistics WHERE table_schema = 'hops' AND table_name = 'hdfs_retry_cache_entry' AND index_name = 'epoch_idx';
+
+SET @create_index_stmt = IF(@index_exists = 0, 'ALTER TABLE hdfs_retry_cache_entry ADD INDEX epoch_idx (epoch);', 'SELECT "Index already exists";');
+PREPARE stmt FROM @create_index_stmt;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;

src/main/java/io/hops/metadata/ndb/dalimpl/hdfs/RetryCacheEntryClusterj.java

@@ -116,13 +116,14 @@ public void prepare(Collection<RetryCacheEntry> removed,
     session.release(changes);
   }
 
-  public int removeOlds(long epoch) throws StorageException{
+  public int removeOlds(long epoch, int batchSize) throws StorageException{
     HopsSession session = connector.obtainSession();
     HopsQueryBuilder qb = session.getQueryBuilder();
     HopsQueryDomainType<RetryCacheEntryDTO> qdt = qb.createQueryDefinition(RetryCacheEntryDTO.class);
     qdt.where(qdt.get("epoch").equal(qdt.param("param")));
     HopsQuery<RetryCacheEntryDTO> query = session.createQuery(qdt);
     query.setParameter("param", epoch);
+    query.setLimits(0, batchSize);
     return query.deletePersistentAll();
   }