Change the separator character for index/type patterns that contain a…

… format.

Colon character gets in the way when some frameworks attempt to fit the input resource
into a Path for HDFS (even though they eventually never use it as an HDFS path). The
break in parsing causes the jobs to fail when using this format.

Applying fix to the format separator in the index pattern.

relates #985

cascading/src/itest/java/org/elasticsearch/hadoop/integration/cascading/AbstractCascadingHadoopJsonSaveTest.java

@@ -72,7 +72,7 @@ public void testIndexPattern() throws Exception {
     @Test
     public void testIndexPatternWithFormat() throws Exception {
         Tap in = sourceTap();
-        Tap out = new EsTap("json-cascading-hadoop-pattern-format-{@timestamp:YYYY-MM-dd}/data", new Fields("line"));
+        Tap out = new EsTap("json-cascading-hadoop-pattern-format-{@timestamp|YYYY-MM-dd}/data", new Fields("line"));
         Pipe pipe = new Pipe("copy");
         build(cfg(), in, out, pipe);
     }

cascading/src/itest/java/org/elasticsearch/hadoop/integration/cascading/AbstractCascadingHadoopSaveTest.java

@@ -116,7 +116,7 @@ public void testIndexPatternWithFormat() throws Exception {
         Properties props = HdpBootstrap.asProperties(CascadingHadoopSuite.configuration);
 
         Tap in = sourceTap();
-        Tap out = new EsTap("cascading-hadoop-pattern-format-{ts:YYYY-MM-dd}/data", new Fields("id", "name", "url", "picture", "ts"));
+        Tap out = new EsTap("cascading-hadoop-pattern-format-{ts|YYYY-MM-dd}/data", new Fields("id", "name", "url", "picture", "ts"));
         Pipe pipe = new Pipe("copy");
         StatsUtils.proxy(new HadoopFlowConnector(props).connect(in, out, pipe)).complete();
     }

cascading/src/itest/java/org/elasticsearch/hadoop/integration/cascading/AbstractCascadingLocalJsonSaveTest.java

@@ -80,7 +80,7 @@ public void testIndexPatternWithFormat() throws Exception {
         properties.put(ConfigurationOptions.ES_INPUT_JSON, "yes");
 
         Tap in = sourceTap();
-        Tap out = new EsTap("json-cascading-local-pattern-format-{@timestamp:YYYY-MM-dd}/data", new Fields("line"));
+        Tap out = new EsTap("json-cascading-local-pattern-format-{@timestamp|YYYY-MM-dd}/data", new Fields("line"));
         Pipe pipe = new Pipe("copy");
         build(properties, in, out, pipe);
     }

cascading/src/itest/java/org/elasticsearch/hadoop/integration/cascading/AbstractCascadingLocalSaveTest.java

@@ -163,7 +163,7 @@ public void testIndexPatternWithFormatAndAlias() throws Exception {
 
         // local file-system source
         Tap in = sourceTap();
-        Tap out = new EsTap("cascading-local-pattern-format-{ts:YYYY-MM-dd}/data", new Fields("id", "name", "url", "picture", "ts"));
+        Tap out = new EsTap("cascading-local-pattern-format-{ts|YYYY-MM-dd}/data", new Fields("id", "name", "url", "picture", "ts"));
         Pipe pipe = new Pipe("copy");
 
         build(properties, in, out, pipe);

mr/src/itest/java/org/elasticsearch/hadoop/integration/mr/AbstractMRNewApiSaveTest.java

@@ -369,7 +369,7 @@ public void testIndexPattern() throws Exception {
     @Test
     public void testIndexPatternWithFormatting() throws Exception {
         Configuration conf = createConf();
-        conf.set(ConfigurationOptions.ES_RESOURCE, "mrnewapi-pattern-format-{@timestamp:YYYY-MM-dd}/data");
+        conf.set(ConfigurationOptions.ES_RESOURCE, "mrnewapi-pattern-format-{@timestamp|YYYY-MM-dd}/data");
         conf.set(ConfigurationOptions.ES_INDEX_AUTO_CREATE, "yes");
 
         runJob(conf);

mr/src/itest/java/org/elasticsearch/hadoop/integration/mr/AbstractMROldApiSaveTest.java

@@ -529,7 +529,7 @@ public void testIndexPattern() throws Exception {
     @Test
     public void testIndexPatternWithFormatting() throws Exception {
         JobConf conf = createJobConf();
-        conf.set(ConfigurationOptions.ES_RESOURCE, "mroldapi-pattern-format-{@timestamp:YYYY-MM-dd}/data");
+        conf.set(ConfigurationOptions.ES_RESOURCE, "mroldapi-pattern-format-{@timestamp|YYYY-MM-dd}/data");
         conf.set(ConfigurationOptions.ES_INDEX_AUTO_CREATE, "yes");
 
         runJob(conf);
@@ -538,7 +538,7 @@ public void testIndexPatternWithFormatting() throws Exception {
     @Test
     public void testIndexPatternWithFormattingAndId() throws Exception {
         JobConf conf = createJobConf();
-        conf.set(ConfigurationOptions.ES_RESOURCE, "mroldapi-pattern-format-{@timestamp:YYYY-MM-dd}-with-id/data");
+        conf.set(ConfigurationOptions.ES_RESOURCE, "mroldapi-pattern-format-{@timestamp|YYYY-MM-dd}-with-id/data");
         conf.set(ConfigurationOptions.ES_MAPPING_ID, "number");
 
         runJob(conf);

mr/src/main/java/org/elasticsearch/hadoop/serialization/field/AbstractIndexExtractor.java

@@ -31,6 +31,8 @@
 
 public abstract class AbstractIndexExtractor implements IndexExtractor, SettingsAware {
 
+    private static final String FORMAT_SEPARATOR = "|";
+
     protected Settings settings;
     protected String pattern;
     protected boolean hasPattern = false;
@@ -65,7 +67,7 @@ protected List<Object> parse(String string) {
             int endPattern = string.indexOf("}");
             Assert.isTrue(endPattern > startPattern + 1, "Invalid pattern given " + string);
             String nestedString = string.substring(startPattern + 1, endPattern);
-            int separator = nestedString.indexOf(":");
+            int separator = nestedString.indexOf(FORMAT_SEPARATOR);
             if (separator > 0) {
                 Assert.isTrue(nestedString.length() > separator + 1, "Invalid format given " + nestedString);
                 String format = nestedString.substring(separator + 1);