Repository: apache/fluo
Issue: FluoOutputFormat is dropping zookeepers from config

Trying to run the stress test on a 20 node aws cluster.  I have 3 zookeepers.  My config looks like the following

```
io.fluo.client.zookeeper.connect=ip-10-1-2-10,ip-10-1-2-11,ip-10-1-2-12/fluo
```

Mappers fail w/ the following exception

```
Caused by: java.lang.IllegalStateException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /config/accumulo.instance.name
    at io.fluo.core.impl.Environment.init(Environment.java:118)
    at io.fluo.core.impl.Environment.<init>(Environment.java:95)
    at io.fluo.core.client.FluoClientImpl.<init>(FluoClientImpl.java:53)
    ... 15 more
```

In the mapper log, I found the following.   So I think something is dropping everything after the 1st comma.

```
2014-11-12 20:39:25,294 INFO [main] org.apache.zookeeper.ZooKeeper: Initiating client connection, connectString=ip-10-1-2-10 sessionTimeout=30000 watcher=org.apache.curator.ConnectionState@60a46327
```


Changed files:
modules/api/src/main/java/io/fluo/api/config/FluoConfiguration.java, modules/api/src/test/java/io/fluo/api/config/FluoConfigurationTest.java

diff --git a/modules/api/src/main/java/io/fluo/api/config/FluoConfiguration.java b/modules/api/src/main/java/io/fluo/api/config/FluoConfiguration.java
index 2d8bd37d..156c38bf 100644
--- a/modules/api/src/main/java/io/fluo/api/config/FluoConfiguration.java
+++ b/modules/api/src/main/java/io/fluo/api/config/FluoConfiguration.java
@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
 import javax.xml.bind.DatatypeConverter;
 
 import com.google.common.base.Charsets;
+import org.apache.commons.configuration.AbstractConfiguration;
 import org.apache.commons.configuration.CompositeConfiguration;
 import org.apache.commons.configuration.Configuration;
 import org.apache.commons.configuration.ConfigurationException;
@@ -124,6 +125,11 @@ public class FluoConfiguration extends CompositeConfiguration {
   
   public FluoConfiguration(Configuration configuration) {
     this();
+    if(configuration instanceof AbstractConfiguration){
+      AbstractConfiguration aconf = (AbstractConfiguration) configuration;
+      aconf.setDelimiterParsingDisabled(true);
+    }
+    
     addConfiguration(configuration);
   }
   
diff --git a/modules/api/src/test/java/io/fluo/api/config/FluoConfigurationTest.java b/modules/api/src/test/java/io/fluo/api/config/FluoConfigurationTest.java
index 94303279..659cfdce 100644
--- a/modules/api/src/test/java/io/fluo/api/config/FluoConfigurationTest.java
+++ b/modules/api/src/test/java/io/fluo/api/config/FluoConfigurationTest.java
@@ -17,7 +17,9 @@ package io.fluo.api.config;
 
 import java.io.File;
 import java.util.NoSuchElementException;
+import java.util.Properties;
 
+import org.apache.commons.configuration.ConfigurationConverter;
 import org.apache.commons.io.IOUtils;
 import org.junit.Assert;
 import org.junit.Test;
@@ -179,4 +181,17 @@ public class FluoConfigurationTest {
     byte[] bytes2 = IOUtils.toByteArray(config.getMetricsYaml());
     Assert.assertArrayEquals(bytes1, bytes2);
   }
+  
+  @Test
+  public void testMultipleZookeepers(){
+    Properties props = new Properties();
+    props.setProperty(FluoConfiguration.CLIENT_ACCUMULO_ZOOKEEPERS_PROP, "zk1,zk2,zk3");
+    props.setProperty(FluoConfiguration.CLIENT_ZOOKEEPER_CONNECT_PROP, "zk1,zk2,zk3/fluo");
+    
+    //ran into a bug where this particular constructor was truncating everything after zk1
+    FluoConfiguration config = new FluoConfiguration(ConfigurationConverter.getConfiguration(props));
+    Assert.assertEquals("zk1,zk2,zk3", config.getAccumuloZookeepers());
+    Assert.assertEquals("zk1,zk2,zk3/fluo", config.getZookeepers());
+  }
+  
 }