memcached cache: switch to AWS elasticache-java-cluster-client and add TLS support

docs/configuration/index.md

@@ -2103,16 +2103,18 @@ In addition to the normal cache metrics, the caffeine cache implementation also
 
 Uses memcached as cache backend. This allows all processes to share the same cache.
 
-|Property|Description|Default|
-|--------|-----------|-------|
-|`druid.cache.expiration`|Memcached [expiration time](https://code.google.com/p/memcached/wiki/NewCommands#Standard_Protocol).|2592000 (30 days)|
-|`druid.cache.timeout`|Maximum time in milliseconds to wait for a response from Memcached.|500|
-|`druid.cache.hosts`|Comma separated list of Memcached hosts `<host:port>`.|none|
-|`druid.cache.maxObjectSize`|Maximum object size in bytes for a Memcached object.|52428800 (50 MiB)|
-|`druid.cache.memcachedPrefix`|Key prefix for all keys in Memcached.|druid|
-|`druid.cache.numConnections`|Number of memcached connections to use.|1|
-|`druid.cache.protocol`|Memcached communication protocol. Can be binary or text.|binary|
-|`druid.cache.locator`|Memcached locator. Can be consistent or array_mod.|consistent|
+| Property                      | Description                                                                                          | Default           |
+|-------------------------------|------------------------------------------------------------------------------------------------------|-------------------|
+| `druid.cache.expiration`      | Memcached [expiration time](https://code.google.com/p/memcached/wiki/NewCommands#Standard_Protocol). | 2592000 (30 days) |
+| `druid.cache.timeout`         | Maximum time in milliseconds to wait for a response from Memcached.                                  | 500               |
+| `druid.cache.hosts`           | Comma separated list of Memcached hosts `<host:port>`.                                               | none              |
+| `druid.cache.maxObjectSize`   | Maximum object size in bytes for a Memcached object.                                                 | 52428800 (50 MiB) |
+| `druid.cache.memcachedPrefix` | Key prefix for all keys in Memcached.                                                                | druid             |
+| `druid.cache.numConnections`  | Number of memcached connections to use.                                                              | 1                 |
+| `druid.cache.protocol`        | Memcached communication protocol. Can be binary or text.                                             | binary            |
+| `druid.cache.locator`         | Memcached locator. Can be consistent or array_mod.                                                   | consistent        |
+| `druid.cache.sslconnection`   | Use TLS based ssl connection for Memcached client.  Boolean                                          | false             |
+| `druid.cache.autodiscovery`   | Use AutoDiscovery feature of AWS Elasticache Memcached. Boolean                                      | false             |
 
 #### Hybrid
 

licenses.yaml

@@ -1638,13 +1638,13 @@ libraries:
 
 ---
 
-name: Spymemcached
+name: aws-elasticache-cluster-client-memcached-for-java
 license_category: binary
 module: java-core
 license_name: Apache License version 2.0
-version: 2.12.3
+version: 1.2.0
 libraries:
-  - net.spy: spymemcached
+  - com.amazonaws: elasticache-java-cluster-client
 
 ---
 

pom.xml

@@ -766,9 +766,9 @@
                 <version>3.3.6</version>
             </dependency>
             <dependency>
-                <groupId>net.spy</groupId>
-                <artifactId>spymemcached</artifactId>
-                <version>2.12.3</version>
+                <groupId>com.amazonaws</groupId>
+                <artifactId>elasticache-java-cluster-client</artifactId>
+                <version>1.2.0</version>
             </dependency>
             <dependency>
                 <groupId>org.antlr</groupId>

server/pom.xml

@@ -134,8 +134,8 @@
             <artifactId>tesla-aether</artifactId>
         </dependency>
         <dependency>
-            <groupId>net.spy</groupId>
-            <artifactId>spymemcached</artifactId>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>elasticache-java-cluster-client</artifactId>
         </dependency>
         <dependency>
             <groupId>org.lz4</groupId>

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

@@ -29,6 +29,12 @@
 import com.google.common.collect.Maps;
 import com.google.common.hash.HashFunction;
 import com.google.common.hash.Hashing;
+import net.spy.memcached.*;
+
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
 import net.spy.memcached.AddrUtil;
 import net.spy.memcached.ConnectionFactory;
 import net.spy.memcached.ConnectionFactoryBuilder;
@@ -52,10 +58,13 @@
 import org.apache.druid.java.util.metrics.AbstractMonitor;
 
 import javax.annotation.Nullable;
+import javax.net.ssl.TrustManagerFactory;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -339,7 +348,7 @@ public void updateHistogram(String name, int amount)
         }
       };
 
-      final ConnectionFactory connectionFactory = new MemcachedCustomConnectionFactoryBuilder()
+      ConnectionFactoryBuilder connectionFactoryBuilder =  new MemcachedCustomConnectionFactoryBuilder()
           // 1000 repetitions gives us good distribution with murmur3_128
           // (approx < 5% difference in counts across nodes, with 5 cache nodes)
           .setKetamaNodeRepetitions(1000)
@@ -355,9 +364,23 @@ public void updateHistogram(String name, int amount)
           .setReadBufferSize(config.getReadBufferSize())
           .setOpQueueFactory(opQueueFactory)
           .setMetricCollector(metricCollector)
-          .setEnableMetrics(MetricType.DEBUG) // Not as scary as it sounds
-          .build();
-
+          .setEnableMetrics(MetricType.DEBUG); // Not as scary as it sounds
+      if(config.usesslconnection()) {
+        // Build SSLContext
+        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+        tmf.init((KeyStore) null);
+        SSLContext sslContext = SSLContext.getInstance("TLS");
+        sslContext.init(null, tmf.getTrustManagers(), null);
+        // Create the client in TLS mode
+        connectionFactoryBuilder.setSSLContext(sslContext);
+      }
+      if(config.autoDiscovery()) {
+        connectionFactoryBuilder.setClientMode(ClientMode.Dynamic);
+      }
+      else {
+        connectionFactoryBuilder.setClientMode(ClientMode.Static);
+      }
+      final ConnectionFactory connectionFactory = connectionFactoryBuilder.build();
       final List<InetSocketAddress> hosts = AddrUtil.getAddresses(config.getHosts());
 
 
@@ -389,7 +412,11 @@ public MemcachedClientIF get()
 
       return new MemcachedCache(clientSupplier, config, monitor);
     }
-    catch (IOException e) {
+    catch (IOException | NoSuchAlgorithmException e) {
+      throw new RuntimeException(e);
+    } catch (KeyStoreException e) {
+      throw new RuntimeException(e);
+    } catch (KeyManagementException e) {
       throw new RuntimeException(e);
     }
   }

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

@@ -63,6 +63,12 @@ public class MemcachedCacheConfig
   @JsonProperty
   private String locator = "consistent";
 
+  @JsonProperty
+  private boolean sslconnection = false;
+
+  @JsonProperty
+  private boolean autodiscovery = false;
+
   public int getExpiration()
   {
     return expiration;
@@ -112,4 +118,12 @@ public String getLocator()
   {
     return locator;
   }
+
+  public boolean usesslconnection() {
+    return sslconnection;
+  }
+
+  public boolean autoDiscovery() {
+    return autodiscovery;
+  }
 }

server/src/test/java/org/apache/druid/client/cache/MemcachedCacheTest.java

@@ -29,14 +29,7 @@
 import com.google.inject.Injector;
 import com.google.inject.Module;
 import com.google.inject.name.Names;
-import net.spy.memcached.BroadcastOpFactory;
-import net.spy.memcached.CASResponse;
-import net.spy.memcached.CASValue;
-import net.spy.memcached.CachedData;
-import net.spy.memcached.ConnectionObserver;
-import net.spy.memcached.MemcachedClientIF;
-import net.spy.memcached.MemcachedNode;
-import net.spy.memcached.NodeLocator;
+import net.spy.memcached.*;
 import net.spy.memcached.internal.BulkFuture;
 import net.spy.memcached.internal.BulkGetCompletionListener;
 import net.spy.memcached.internal.OperationFuture;
@@ -224,6 +217,26 @@ public void emit(Event event)
     }
   }
 
+  @Test
+  public void testSslConnection()
+  {
+    final MemcachedCacheConfig config = new MemcachedCacheConfig()
+    {
+      @Override
+      public boolean usesslconnection()
+      {
+        return true;
+      }
+
+      @Override
+      public String getHosts()
+      {
+        return "localhost:9999";
+      }
+    };
+    MemcachedCache cache = MemcachedCache.create(config);
+  }
+
   @Test
   public void testSanity()
   {
@@ -316,6 +329,12 @@ public Collection<SocketAddress> getAvailableServers()
     throw new UnsupportedOperationException("not implemented");
   }
 
+  @Override
+  public boolean refreshCertificate()
+  {
+    return true;
+  }
+
   @Override
   public Collection<SocketAddress> getUnavailableServers()
   {