acl for zookeeper is added

docs/content/configuration/index.md

@@ -40,6 +40,7 @@ We recommend just setting the base ZK path and the ZK service host, but all ZK p
 |--------|-----------|-------|
 |`druid.zk.service.sessionTimeoutMs`|ZooKeeper session timeout, in milliseconds.|`30000`|
 |`druid.zk.service.compress`|Boolean flag for whether or not created Znodes should be compressed.|`true`|
+|`druid.zk.service.acl`|Boolean flag for whether or not to enable ACL security for ZooKeeper. If ACL is enabled, zNode creators will have all permissions.|`false`|
 
 #### Path Configuration
 Druid interacts with ZK through a set of standard path configurations. We recommend just setting the base ZK path, but all ZK paths that Druid uses can be overwritten to absolute paths.

server/src/main/java/io/druid/curator/CuratorConfig.java

@@ -20,6 +20,7 @@
 package io.druid.curator;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
 
 import javax.validation.constraints.Min;
 
@@ -37,6 +38,9 @@ public class CuratorConfig
   @JsonProperty("compress")
   private boolean enableCompression = true;
 
+  @JsonProperty("acl")
+  private boolean enableAcl = false;
+
   public String getZkHosts()
   {
     return zkHosts;
@@ -57,13 +61,25 @@ public class CuratorConfig
     this.zkSessionTimeoutMs = zkSessionTimeoutMs;
   }
 
-  public Boolean getEnableCompression()
+  public boolean getEnableCompression()
   {
     return enableCompression;
   }
 
   public void setEnableCompression(Boolean enableCompression)
   {
+    Preconditions.checkNotNull(enableCompression, "enableCompression");
     this.enableCompression = enableCompression;
   }
+
+  public boolean getEnableAcl()
+  {
+    return enableAcl;
+  }
+
+  public void setEnableAcl(Boolean enableAcl)
+  {
+    Preconditions.checkNotNull(enableAcl, "enableAcl");
+    this.enableAcl = enableAcl;
+  }
 }

server/src/main/java/io/druid/curator/CuratorModule.java

@@ -24,14 +24,23 @@ import com.google.inject.Module;
 import com.google.inject.Provides;
 import com.metamx.common.lifecycle.Lifecycle;
 import com.metamx.common.logger.Logger;
+
 import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.LazySingleton;
+
+import org.apache.curator.framework.api.ACLProvider;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.DefaultACLProvider;
 import org.apache.curator.retry.BoundedExponentialBackoffRetry;
 
 import java.io.IOException;
 
+import java.util.List;
+
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+
 /**
  */
 public class CuratorModule implements Module
@@ -43,19 +52,22 @@ public class CuratorModule implements Module
   {
     JsonConfigProvider.bind(
         binder, "druid.zk.service",
-        CuratorConfig.class);
+        CuratorConfig.class
+    );
   }
 
-  @Provides @LazySingleton
+  @Provides
+  @LazySingleton
   public CuratorFramework makeCurator(CuratorConfig config, Lifecycle lifecycle) throws IOException
   {
     final CuratorFramework framework =
         CuratorFrameworkFactory.builder()
                                .connectString(config.getZkHosts())
                                .sessionTimeoutMs(config.getZkSessionTimeoutMs())
-            .retryPolicy(new BoundedExponentialBackoffRetry(1000, 45000, 30))
-            .compressionProvider(new PotentiallyGzippedCompressionProvider(config.getEnableCompression()))
-            .build();
+                               .retryPolicy(new BoundedExponentialBackoffRetry(1000, 45000, 30))
+                               .compressionProvider(new PotentiallyGzippedCompressionProvider(config.getEnableCompression()))
+                               .aclProvider(config.getEnableAcl() ? new SecuredACLProvider() : new DefaultACLProvider())
+                               .build();
 
     lifecycle.addHandler(
         new Lifecycle.Handler()
@@ -78,4 +90,19 @@ public class CuratorModule implements Module
 
     return framework;
   }
+
+  class SecuredACLProvider implements ACLProvider
+  {
+    @Override
+    public List<ACL> getDefaultAcl()
+    {
+      return ZooDefs.Ids.CREATOR_ALL_ACL;
+    }
+
+    @Override
+    public List<ACL> getAclForPath(String path)
+    {
+      return ZooDefs.Ids.CREATOR_ALL_ACL;
+    }
+  }
 }

server/src/test/java/io/druid/curator/CuratorConfigTest.java

@@ -28,12 +28,14 @@ import java.lang.reflect.InvocationTargetException;
 public class CuratorConfigTest extends JsonConfigTesterBase<CuratorConfig>
 {
   @Test
-  public void testHostName() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException
+  public void testSerde() throws IllegalAccessException, NoSuchMethodException, InvocationTargetException
   {
-    propertyValues.put(getPropertyKey("host"),"fooHost");
+    propertyValues.put(getPropertyKey("host"), "fooHost");
+    propertyValues.put(getPropertyKey("acl"), "true");
     testProperties.putAll(propertyValues);
     configProvider.inject(testProperties, configurator);
     CuratorConfig config = configProvider.get().get();
     Assert.assertEquals("fooHost", config.getZkHosts());
+    Assert.assertEquals(true, config.getEnableAcl());
   }
 }