add close method in Cache interface (#6540)

* add close method in Cache interface

* address comments

* address comments and fix travis-ci

* use try-finally

extensions-contrib/redis-cache/src/main/java/org/apache/druid/client/cache/RedisCache.java

@@ -21,6 +21,7 @@ package org.apache.druid.client.cache;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
@@ -29,6 +30,7 @@ import redis.clients.jedis.JedisPool;
 import redis.clients.jedis.JedisPoolConfig;
 import redis.clients.jedis.exceptions.JedisException;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -147,6 +149,13 @@ public class RedisCache implements Cache
     // no resources to cleanup
   }
 
+  @Override
+  @LifecycleStop
+  public void close() throws IOException
+  {
+    pool.close();
+  }
+
   @Override
   public CacheStats getStats()
   {

server/src/main/java/org/apache/druid/client/cache/Cache.java

@@ -24,13 +24,14 @@ import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 
 import javax.annotation.Nullable;
+import java.io.Closeable;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Map;
 
 /**
  */
-public interface Cache
+public interface Cache extends Closeable
 {
   @Nullable
   byte[] get(NamedKey key);

server/src/main/java/org/apache/druid/client/cache/CaffeineCache.java

@@ -28,11 +28,13 @@ import com.google.common.collect.Maps;
 import net.jpountz.lz4.LZ4Compressor;
 import net.jpountz.lz4.LZ4Factory;
 import net.jpountz.lz4.LZ4FastDecompressor;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
 import org.apache.druid.utils.JvmUtils;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.OptionalLong;
@@ -115,6 +117,13 @@ public class CaffeineCache implements org.apache.druid.client.cache.Cache
     }
   }
 
+  @Override
+  @LifecycleStop
+  public void close() throws IOException
+  {
+    cache.cleanUp();
+  }
+
   @Override
   public org.apache.druid.client.cache.CacheStats getStats()
   {

server/src/main/java/org/apache/druid/client/cache/HybridCache.java

@@ -20,10 +20,13 @@
 package org.apache.druid.client.cache;
 
 import com.google.common.collect.Sets;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 
 import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -127,8 +130,35 @@ public class HybridCache implements Cache
   @Override
   public void close(String namespace)
   {
-    level1.close(namespace);
-    level2.close(namespace);
+    Throwable t = null;
+    try {
+      level1.close(namespace);
+    }
+    catch (Throwable t1) {
+      t = t1;
+      throw t1;
+    }
+    finally {
+      if (t != null) {
+        try {
+          level2.close(namespace);
+        }
+        catch (Throwable t2) {
+          t.addSuppressed(t2);
+        }
+      } else {
+        level2.close(namespace);
+      }
+    }
+  }
+
+  @Override
+  @LifecycleStop
+  public void close() throws IOException
+  {
+    try (Closeable closeable1 = level1; Closeable closeable2 = level2) {
+      // Just for closing
+    }
   }
 
   @Override

server/src/main/java/org/apache/druid/client/cache/MapCache.java

@@ -20,8 +20,10 @@
 package org.apache.druid.client.cache;
 
 import com.google.common.primitives.Ints;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -144,6 +146,15 @@ public class MapCache implements Cache
     }
   }
 
+  @Override
+  @LifecycleStop
+  public void close() throws IOException
+  {
+    baseMap.clear();
+    byteCountingLRUMap.clear();
+    namespaceId.clear();
+  }
+
   private byte[] getNamespaceId(final String identifier)
   {
     synchronized (namespaceId) {

server/src/main/java/org/apache/druid/client/cache/MemcachedCache.java

@@ -46,6 +46,7 @@ import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.druid.collections.ResourceHolder;
 import org.apache.druid.collections.StupidResourceHolder;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
@@ -596,6 +597,13 @@ public class MemcachedCache implements Cache
     // no resources to cleanup
   }
 
+  @Override
+  @LifecycleStop
+  public void close() throws IOException
+  {
+    monitor.stop();
+  }
+
   public static final int MAX_PREFIX_LENGTH =
       MemcachedClientIF.MAX_KEY_LENGTH
       - 40 // length of namespace hash

server/src/test/java/org/apache/druid/client/CachingQueryRunnerTest.java

@@ -253,6 +253,11 @@ public class CachingQueryRunnerTest
       {
       }
 
+      @Override
+      public void close() throws IOException
+      {
+      }
+
       @Override
       public CacheStats getStats()
       {