SOLR-6089: When using the HDFS block cache, when a file is deleted, it's underlying data entries in the block cache are not removed, which is a problem with the global block cache option.

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1619427 13f79535-47bb-0310-9956-ffa450edef68

solr/CHANGES.txt

@@ -302,6 +302,10 @@ Bug Fixes
 * SOLR-6268: HdfsUpdateLog has a race condition that can expose a closed HDFS FileSystem instance and should 
   close it's FileSystem instance if either inherited close method is called. (Mark Miller)
 
+* SOLR-6089: When using the HDFS block cache, when a file is deleted, it's underlying data entries in the 
+  block cache are not removed, which is a problem with the global block cache option. 
+  (Mark Miller, Patrick Hunt)
+
 Optimizations
 ---------------------
 

solr/contrib/map-reduce/src/test/org/apache/solr/hadoop/MorphlineGoLiveMiniMRTest.java

@@ -44,6 +44,7 @@ import org.apache.hadoop.util.JarFinder;
 import org.apache.hadoop.util.ToolRunner;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.lucene.util.Constants;
+import org.apache.lucene.util.LuceneTestCase;
 import org.apache.lucene.util.LuceneTestCase.Slow;
 import org.apache.lucene.util.LuceneTestCase.SuppressCodecs;
 import org.apache.solr.SolrTestCaseJ4.SuppressSSL;
@@ -65,6 +66,7 @@ import org.apache.solr.common.cloud.Slice;
 import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.solr.common.cloud.ZkCoreNodeProps;
 import org.apache.solr.common.params.CollectionParams.CollectionAction;
+import org.apache.solr.common.params.CoreAdminParams;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.util.NamedList;
 import org.apache.solr.hadoop.hack.MiniMRClientCluster;
@@ -128,6 +130,10 @@ public class MorphlineGoLiveMiniMRTest extends AbstractFullDistribZkTestBase {
   
   @BeforeClass
   public static void setupClass() throws Exception {
+    System.setProperty("solr.hdfs.blockcache.global", Boolean.toString(LuceneTestCase.random().nextBoolean()));
+    System.setProperty("solr.hdfs.blockcache.enabled", Boolean.toString(LuceneTestCase.random().nextBoolean()));
+    System.setProperty("solr.hdfs.blockcache.blocksperbank", "2048");
+    
     solrHomeDirectory = createTempDir();
     assumeTrue(
             "Currently this test can only be run without the lucene test security policy in place",
@@ -156,8 +162,6 @@ public class MorphlineGoLiveMiniMRTest extends AbstractFullDistribZkTestBase {
     int taskTrackers = 2;
     int dataNodes = 2;
     
-    System.setProperty("solr.hdfs.blockcache.enabled", "false");
-    
     JobConf conf = new JobConf();
     conf.set("dfs.block.access.token.enable", "false");
     conf.set("dfs.permissions", "true");
@@ -218,6 +222,8 @@ public class MorphlineGoLiveMiniMRTest extends AbstractFullDistribZkTestBase {
   
   @AfterClass
   public static void teardownClass() throws Exception {
+    System.clearProperty("solr.hdfs.blockcache.global");
+    System.clearProperty("solr.hdfs.blockcache.blocksperbank");
     System.clearProperty("solr.hdfs.blockcache.enabled");
     System.clearProperty("hadoop.log.dir");
     System.clearProperty("test.build.dir");
@@ -629,6 +635,63 @@ public class MorphlineGoLiveMiniMRTest extends AbstractFullDistribZkTestBase {
       assertEquals(RECORD_COUNT, executeSolrQuery(cloudClient, "*:*").size());
     }
     
+    // delete collection
+    ModifiableSolrParams params = new ModifiableSolrParams();
+    params.set("action", CollectionAction.DELETE.toString());
+    params.set(CoreAdminParams.DELETE_INSTANCE_DIR, true);
+    params.set(CoreAdminParams.DELETE_DATA_DIR, true);
+    params.set(CoreAdminParams.DELETE_INDEX, true);
+    params.set("name", replicatedCollection);
+    QueryRequest request = new QueryRequest(params);
+    request.setPath("/admin/collections");
+    cloudClient.request(request);
+
+    
+    long timeout = System.currentTimeMillis() + 10000;
+    while (cloudClient.getZkStateReader().getClusterState().hasCollection(replicatedCollection)) {
+      if (System.currentTimeMillis() > timeout) {
+        throw new AssertionError("Timeout waiting to see removed collection leave clusterstate");
+      }
+      
+      Thread.sleep(200);
+      cloudClient.getZkStateReader().updateClusterState(true);
+    }
+    
+    if (TEST_NIGHTLY) {
+      createCollection(replicatedCollection, 11, 3, 11);
+    } else {
+      createCollection(replicatedCollection, 2, 3, 2);
+    }
+    
+    waitForRecoveriesToFinish(replicatedCollection, false);
+    printLayout();
+    assertEquals(0, executeSolrQuery(cloudClient, "*:*").getNumFound());
+    
+    
+    args = new String[] {
+        "--solr-home-dir=" + MINIMR_CONF_DIR.getAbsolutePath(),
+        "--output-dir=" + outDir.toString(),
+        "--shards", "2",
+        "--mappers=3",
+        "--verbose",
+        "--go-live", 
+        "--go-live-threads", Integer.toString(random().nextInt(15) + 1),  dataDir.toString()
+    };
+    args = prependInitialArgs(args);
+
+    argList = new ArrayList<>();
+    getShardUrlArgs(argList, replicatedCollection);
+    args = concat(args, argList.toArray(new String[0]));
+    
+    tool = new MapReduceIndexerTool();
+    res = ToolRunner.run(jobConf, tool, args);
+    assertEquals(0, res);
+    assertTrue(tool.job.isComplete());
+    assertTrue(tool.job.isSuccessful());
+    
+    checkConsistency(replicatedCollection);
+    
+    assertEquals(RECORD_COUNT, executeSolrQuery(cloudClient, "*:*").size());
   }
 
   private void getShardUrlArgs(List<String> args) {

solr/contrib/morphlines-core/src/test-files/solr/minimr/conf/solrconfig.xml

@@ -130,6 +130,7 @@
     <bool name="solr.hdfs.nrtcachingdirectory.enable">${solr.hdfs.nrtcachingdirectory.enable:true}</bool>
     <int name="solr.hdfs.nrtcachingdirectory.maxmergesizemb">${solr.hdfs.nrtcachingdirectory.maxmergesizemb:16}</int>
     <int name="solr.hdfs.nrtcachingdirectory.maxcachedmb">${solr.hdfs.nrtcachingdirectory.maxcachedmb:192}</int>
+    <str name="solr.hdfs.blockcache.global">${solr.hdfs.blockcache.global:false}</str>
   </directoryFactory> 
 
   <!-- The CodecFactory for defining the format of the inverted index.

solr/contrib/morphlines-core/src/test-files/solr/solrcloud/conf/solrconfig.xml

@@ -133,6 +133,7 @@
     <bool name="solr.hdfs.nrtcachingdirectory.enable">${solr.hdfs.nrtcachingdirectory.enable:true}</bool>
     <int name="solr.hdfs.nrtcachingdirectory.maxmergesizemb">${solr.hdfs.nrtcachingdirectory.maxmergesizemb:16}</int>
     <int name="solr.hdfs.nrtcachingdirectory.maxcachedmb">${solr.hdfs.nrtcachingdirectory.maxcachedmb:192}</int>
+    <str name="solr.hdfs.blockcache.global">${solr.hdfs.blockcache.global:false}</str>
   </directoryFactory> 
 
   <!-- The CodecFactory for defining the format of the inverted index.

solr/core/src/java/org/apache/solr/core/HdfsDirectoryFactory.java

@@ -139,11 +139,11 @@ public class HdfsDirectoryFactory extends CachingDirectoryFactory {
       int bufferSize = params.getInt("solr.hdfs.blockcache.bufferstore.buffersize", 128);
       int bufferCount = params.getInt("solr.hdfs.blockcache.bufferstore.buffercount", 128 * 128);
       
-      BlockCache blockCache = getBlockDirectoryCache(path, numberOfBlocksPerBank,
+      BlockCache blockCache = getBlockDirectoryCache(numberOfBlocksPerBank,
           blockSize, bankCount, directAllocation, slabSize,
           bufferSize, bufferCount, blockCacheGlobal);
       
-      Cache cache = new BlockDirectoryCache(blockCache, path, metrics);
+      Cache cache = new BlockDirectoryCache(blockCache, path, metrics, blockCacheGlobal);
       HdfsDirectory hdfsDirectory = new HdfsDirectory(new Path(path), conf);
       dir = new BlockDirectory(path, hdfsDirectory, cache, null,
           blockCacheReadEnabled, blockCacheWriteEnabled);
@@ -164,17 +164,16 @@ public class HdfsDirectoryFactory extends CachingDirectoryFactory {
     return dir;
   }
 
-  private BlockCache getBlockDirectoryCache(String path,
+  private BlockCache getBlockDirectoryCache(int numberOfBlocksPerBank, int blockSize, int bankCount,
-      int numberOfBlocksPerBank, int blockSize, int bankCount,
       boolean directAllocation, int slabSize, int bufferSize, int bufferCount, boolean staticBlockCache) {
     if (!staticBlockCache) {
       LOG.info("Creating new single instance HDFS BlockCache");
       return createBlockCache(numberOfBlocksPerBank, blockSize, bankCount, directAllocation, slabSize, bufferSize, bufferCount);
     }
-    LOG.info("Creating new global HDFS BlockCache");
     synchronized (HdfsDirectoryFactory.class) {
       
       if (globalBlockCache == null) {
+        LOG.info("Creating new global HDFS BlockCache");
         globalBlockCache = createBlockCache(numberOfBlocksPerBank, blockSize, bankCount,
             directAllocation, slabSize, bufferSize, bufferCount);
       }

solr/core/src/java/org/apache/solr/store/blockcache/BlockCache.java

@@ -80,6 +80,10 @@ public class BlockCache {
     this.blockSize = blockSize;
   }
   
+  public void release(BlockCacheKey key) {
+    releaseLocation(cache.get(key));
+  }
+  
   private void releaseLocation(BlockCacheLocation location) {
     if (location == null) {
       return;

solr/core/src/java/org/apache/solr/store/blockcache/BlockDirectory.java

@@ -80,19 +80,27 @@ public class BlockDirectory extends Directory {
     
     @Override
     public void renameCacheFile(String source, String dest) {}
+
+    @Override
+    public void releaseResources() {}
   };
   
-  private Directory directory;
+  private final Directory directory;
-  private int blockSize;
+  private final int blockSize;
-  private String dirName;
+  private final String dirName;
   private final Cache cache;
-  private Set<String> blockCacheFileTypes;
+  private final Set<String> blockCacheFileTypes;
   private final boolean blockCacheReadEnabled;
   private final boolean blockCacheWriteEnabled;
 
   public BlockDirectory(String dirName, Directory directory, Cache cache,
       Set<String> blockCacheFileTypes, boolean blockCacheReadEnabled,
       boolean blockCacheWriteEnabled) throws IOException {
+    this(dirName, directory, cache, blockCacheFileTypes, blockCacheReadEnabled, blockCacheWriteEnabled, false);
+  }
+  public BlockDirectory(String dirName, Directory directory, Cache cache,
+      Set<String> blockCacheFileTypes, boolean blockCacheReadEnabled,
+      boolean blockCacheWriteEnabled, boolean releaseBlocksOnClose) throws IOException {
     this.dirName = dirName;
     this.directory = directory;
     blockSize = BLOCK_SIZE;
@@ -233,7 +241,10 @@ public class BlockDirectory extends Directory {
       for (String file : files) {
         cache.delete(getFileCacheName(file));
       }
+      // segments.gen won't be removed above
+      cache.delete(dirName + "/" + "segments.gen");
       
+      cache.releaseResources();
     } catch (FileNotFoundException e) {
       // the local file system folder may be gone
     } finally {

solr/core/src/java/org/apache/solr/store/blockcache/BlockDirectoryCache.java

@@ -17,7 +17,10 @@ package org.apache.solr.store.blockcache;
  * limitations under the License.
  */
 
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -26,15 +29,23 @@ import java.util.concurrent.atomic.AtomicInteger;
  */
 public class BlockDirectoryCache implements Cache {
   private final BlockCache blockCache;
-  private AtomicInteger counter = new AtomicInteger();
+  private final AtomicInteger counter = new AtomicInteger();
-  private Map<String,Integer> names = new ConcurrentHashMap<>();
+  private final Map<String,Integer> names = new ConcurrentHashMap<>();
-  private String path;
+  private Set<BlockCacheKey> keys;
-  private Metrics metrics;
+  private final String path;
+  private final Metrics metrics;
   
   public BlockDirectoryCache(BlockCache blockCache, String path, Metrics metrics) {
+    this(blockCache, path, metrics, false);
+  }
+  
+  public BlockDirectoryCache(BlockCache blockCache, String path, Metrics metrics, boolean releaseBlocks) {
     this.blockCache = blockCache;
     this.path = path;
     this.metrics = metrics;
+    if (releaseBlocks) {
+      keys = Collections.synchronizedSet(new HashSet<BlockCacheKey>());
+    }
   }
   
   /**
@@ -64,6 +75,9 @@ public class BlockDirectoryCache implements Cache {
     blockCacheKey.setBlock(blockId);
     blockCacheKey.setFile(file);
     blockCache.store(blockCacheKey, blockOffset, buffer, offset, length);
+    if (keys != null) {
+      keys.add(blockCacheKey);
+    }
   }
   
   @Override
@@ -100,4 +114,13 @@ public class BlockDirectoryCache implements Cache {
       names.put(dest, file);
     }
   }
+
+  @Override
+  public void releaseResources() {
+    if (keys != null) {
+      for (BlockCacheKey key : keys) {
+        blockCache.release(key);
+      }
+    }
+  }
 }

solr/core/src/java/org/apache/solr/store/blockcache/Cache.java

@@ -62,4 +62,9 @@ public interface Cache {
    */
   void renameCacheFile(String source, String dest);
 
+  /**
+   * Release any resources associated with the cache.
+   */
+  void releaseResources();
+  
 }

solr/core/src/test/org/apache/solr/cloud/ChaosMonkeySafeLeaderTest.java

@@ -173,14 +173,6 @@ public class ChaosMonkeySafeLeaderTest extends AbstractFullDistribZkTestBase {
     checkForCollection("testcollection",numShardsNumReplicas, null);
   }
   
-  private void randomlyEnableAutoSoftCommit() {
-    if (r.nextBoolean()) {
-      enableAutoSoftCommit(1000);
-    } else {
-      log.info("Not turning on auto soft commit");
-    }
-  }
-  
   // skip the randoms - they can deadlock...
   @Override
   protected void indexr(Object... fields) throws Exception {

solr/core/src/test/org/apache/solr/cloud/hdfs/HdfsTestUtil.java

@@ -13,7 +13,6 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
 import org.apache.lucene.util.LuceneTestCase;
-import org.apache.lucene.util.TestUtil;
 import org.apache.solr.SolrTestCaseJ4;
 
 /*

solr/core/src/test/org/apache/solr/cloud/hdfs/StressHdfsTest.java

@@ -90,6 +90,8 @@ public class StressHdfsTest extends BasicDistributedZkTest {
   
   @Override
   public void doTest() throws Exception {
+    randomlyEnableAutoSoftCommit();
+    
     int cnt = random().nextInt(2) + 1;
     for (int i = 0; i < cnt; i++) {
       createAndDeleteCollection();
@@ -161,19 +163,19 @@ public class StressHdfsTest extends BasicDistributedZkTest {
     
     int i = 0;
     for (SolrServer client : clients) {
-      HttpSolrServer c = new HttpSolrServer(getBaseUrl(client)
+      HttpSolrServer c = new HttpSolrServer(getBaseUrl(client) + "/" + DELETE_DATA_DIR_COLLECTION);
-          + "/delete_data_dir");
       try {
-        c.add(getDoc("id", i++));
+        int docCnt = random().nextInt(1000) + 1;
-        if (random().nextBoolean()) c.add(getDoc("id", i++));
+        for (int j = 0; j < docCnt; j++) {
-        if (random().nextBoolean()) c.add(getDoc("id", i++));
+          c.add(getDoc("id", i++, "txt_t", "just some random text for a doc"));
+        }
+
         if (random().nextBoolean()) {
           c.commit();
         } else {
           c.commit(true, true, true);
         }
         
-        c.query(new SolrQuery("id:" + i));
         c.setConnectionTimeout(30000);
         NamedList<Object> response = c.query(
             new SolrQuery().setRequestHandler("/admin/system")).getResponse();
@@ -192,6 +194,10 @@ public class StressHdfsTest extends BasicDistributedZkTest {
       assertEquals(0, cloudClient.query(new SolrQuery("*:*")).getResults().getNumFound());
     }
     
+    cloudClient.commit();
+    cloudClient.query(new SolrQuery("*:*"));
+    
+    // delete collection
     ModifiableSolrParams params = new ModifiableSolrParams();
     params.set("action", CollectionAction.DELETE.toString());
     params.set("name", DELETE_DATA_DIR_COLLECTION);

solr/core/src/test/org/apache/solr/store/blockcache/BlockDirectoryTest.java

@@ -22,17 +22,13 @@ import java.io.IOException;
 import java.util.Map;
 import java.util.Random;
 
-import org.apache.commons.io.FileUtils;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.store.MergeInfo;
-import org.apache.lucene.util.LuceneTestCase;
-
 import org.apache.solr.SolrTestCaseJ4;
-import org.apache.solr.store.hdfs.HdfsDirectory;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -89,6 +85,9 @@ public class BlockDirectoryTest extends SolrTestCaseJ4 {
     @Override
     public void renameCacheFile(String source, String dest) {
     }
+
+    @Override
+    public void releaseResources() {}
   }
 
   private static final int MAX_NUMBER_OF_WRITES = 10000;

solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java

@@ -1092,6 +1092,14 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
     }
   }
   
+  protected void randomlyEnableAutoSoftCommit() {
+    if (r.nextBoolean()) {
+      enableAutoSoftCommit(1000);
+    } else {
+      log.info("Not turning on auto soft commit");
+    }
+  }
+  
   protected void enableAutoSoftCommit(int time) {
     log.info("Turning on auto soft commit: " + time);
     for (List<CloudJettyRunner> jettyList : shardToJetty.values()) {