Improve support of old/ancient Jackson version (1.5)

Backport ObjectReader from Jackson to 1.5 for easy, non-initializing
parsing of the json doc

fix #215

build.gradle

@@ -105,7 +105,7 @@ if (rootProject == project) {
 
 configurations.all {
   resolutionStrategy {
-	forcedModules = ['commons-httpclient:commons-httpclient:3.0.1']
+	forcedModules = ['commons-httpclient:commons-httpclient:3.0.1', "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion", "org.codehaus.jackson:jackson-core-asl:$jacksonVersion"]
   }
 }
 

gradle.properties

@@ -15,7 +15,7 @@ pigVersion = 0.12.1
 # pig 0.11 depends on Joda but doens't declare it in its pom
 jodaVersion = 1.6
 # note the versions here are tied to the ones in Hadoop distro - 1.8.8
-jacksonVersion = 1.8.8
+jacksonVersion = 1.5.2
 cascadingVersion = 2.5.4
 luceneVersion = 4.8.1
 

mr/src/main/java/org/elasticsearch/hadoop/rest/NetworkClient.java

@@ -71,7 +71,7 @@ private boolean selectNextNode() {
         return true;
     }
 
-    public Response execute(Request request) throws EsHadoopInvalidRequest {
+    public Response execute(Request request) {
         Response response = null;
 
         boolean newNode;

mr/src/main/java/org/elasticsearch/hadoop/rest/RestClient.java

@@ -32,14 +32,14 @@
 
 import org.codehaus.jackson.JsonParser;
 import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.map.ObjectReader;
 import org.elasticsearch.hadoop.cfg.ConfigurationOptions;
 import org.elasticsearch.hadoop.cfg.Settings;
 import org.elasticsearch.hadoop.rest.Request.Method;
 import org.elasticsearch.hadoop.rest.stats.Stats;
 import org.elasticsearch.hadoop.rest.stats.StatsAware;
 import org.elasticsearch.hadoop.serialization.ParsingUtils;
 import org.elasticsearch.hadoop.serialization.dto.Node;
+import org.elasticsearch.hadoop.serialization.json.BackportedObjectReader;
 import org.elasticsearch.hadoop.serialization.json.JacksonJsonParser;
 import org.elasticsearch.hadoop.util.ByteSequence;
 import org.elasticsearch.hadoop.util.BytesArray;
@@ -160,7 +160,7 @@ public void bulk(Resource resource, TrackingBytesArray data) {
     @SuppressWarnings("rawtypes")
     private boolean retryFailedEntries(InputStream content, TrackingBytesArray data) {
         try {
-            ObjectReader r = mapper.reader(Map.class);
+            BackportedObjectReader r = BackportedObjectReader.create(mapper, Map.class);
             JsonParser parser = mapper.getJsonFactory().createJsonParser(content);
             try {
                 if (ParsingUtils.seek("items", new JacksonJsonParser(parser)) == null) {

mr/src/main/java/org/elasticsearch/hadoop/serialization/json/BackportedJacksonMappingIterator.java

@@ -0,0 +1,147 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.hadoop.serialization.json;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.codehaus.jackson.JsonParser;
+import org.codehaus.jackson.JsonStreamContext;
+import org.codehaus.jackson.JsonToken;
+import org.codehaus.jackson.map.DeserializationContext;
+import org.codehaus.jackson.map.JsonDeserializer;
+import org.codehaus.jackson.map.JsonMappingException;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.RuntimeJsonMappingException;
+import org.codehaus.jackson.type.JavaType;
+
+/**
+ * Backported class from Jackson 1.8.8 for Jackson 1.5.2
+ *
+ * Iterator exposed by {@link ObjectMapper} when binding sequence of
+ * objects. Extension is done to allow more convenient exposing of
+ * {@link IOException} (which basic {@link Iterator} does not expose)
+ *
+ * @since 1.8
+ */
+public class BackportedJacksonMappingIterator<T> implements Iterator<T> {
+    protected final static BackportedJacksonMappingIterator<?> EMPTY_ITERATOR = new BackportedJacksonMappingIterator<Object>(null, null, null, null);
+
+    protected final JavaType _type;
+
+    protected final DeserializationContext _context;
+
+    protected final JsonDeserializer<T> _deserializer;
+
+    protected final JsonParser _parser;
+
+    @SuppressWarnings("unchecked")
+    protected BackportedJacksonMappingIterator(JavaType type, JsonParser jp, DeserializationContext ctxt, JsonDeserializer<?> deser) {
+        _type = type;
+        _parser = jp;
+        _context = ctxt;
+        _deserializer = (JsonDeserializer<T>) deser;
+
+        /* One more thing: if we are at START_ARRAY (but NOT root-level
+         * one!), advance to next token (to allow matching END_ARRAY)
+         */
+        if (jp != null && jp.getCurrentToken() == JsonToken.START_ARRAY) {
+            JsonStreamContext sc = jp.getParsingContext();
+            // safest way to skip current token is to clear it (so we'll advance soon)
+            if (!sc.inRoot()) {
+                jp.clearCurrentToken();
+            }
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static <T> BackportedJacksonMappingIterator<T> emptyIterator() {
+        return (BackportedJacksonMappingIterator<T>) EMPTY_ITERATOR;
+    }
+
+    /*
+    /**********************************************************
+    /* Basic iterator impl
+    /**********************************************************
+     */
+
+    @Override
+    public boolean hasNext() {
+        try {
+            return hasNextValue();
+        } catch (JsonMappingException e) {
+            throw new RuntimeJsonMappingException(e.getMessage(), e);
+        } catch (IOException e) {
+            throw new RuntimeException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public T next() {
+        try {
+            return nextValue();
+        } catch (JsonMappingException e) {
+            throw new RuntimeJsonMappingException(e.getMessage(), e);
+        } catch (IOException e) {
+            throw new RuntimeException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void remove() {
+        throw new UnsupportedOperationException();
+    }
+
+    /*
+    /**********************************************************
+    /* Extended API
+    /**********************************************************
+     */
+
+    /**
+     * Equivalent of {@link #next} but one that may throw checked
+     * exceptions from Jackson due to invalid input.
+     */
+    public boolean hasNextValue() throws IOException {
+        if (_parser == null) {
+            return false;
+        }
+        JsonToken t = _parser.getCurrentToken();
+        if (t == null) { // un-initialized or cleared; find next
+            t = _parser.nextToken();
+            // If EOF, no more
+            if (t == null) {
+                _parser.close();
+                return false;
+            }
+            // And similarly if we hit END_ARRAY; except that we won't close parser
+            if (t == JsonToken.END_ARRAY) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public T nextValue() throws IOException {
+        T result = _deserializer.deserialize(_parser, _context);
+        // Need to consume the token too
+        _parser.clearCurrentToken();
+        return result;
+    }
+}

mr/src/main/java/org/elasticsearch/hadoop/serialization/json/BackportedObjectReader.java

@@ -0,0 +1,180 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.hadoop.serialization.json;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.JsonGenerator;
+import org.codehaus.jackson.JsonParser;
+import org.codehaus.jackson.JsonProcessingException;
+import org.codehaus.jackson.map.DeserializationConfig;
+import org.codehaus.jackson.map.DeserializationContext;
+import org.codehaus.jackson.map.DeserializerProvider;
+import org.codehaus.jackson.map.JsonDeserializer;
+import org.codehaus.jackson.map.JsonMappingException;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.deser.StdDeserializationContext;
+import org.codehaus.jackson.map.introspect.VisibilityChecker;
+import org.codehaus.jackson.map.jsontype.TypeResolverBuilder;
+import org.codehaus.jackson.map.type.TypeFactory;
+import org.codehaus.jackson.type.JavaType;
+import org.eclipse.jdt.internal.core.Assert;
+import org.elasticsearch.hadoop.util.ReflectionUtils;
+
+/**
+ * Backported class from Jackson 1.8.8 for Jackson 1.5.2
+ *
+ * Builder object that can be used for per-serialization configuration of
+ * deserialization parameters, such as root type to use or object
+ * to update (instead of constructing new instance).
+ * Uses "fluid" (aka builder) pattern so that instances are immutable
+ * (and thus fully thread-safe with no external synchronization);
+ * new instances are constructed for different configurations.
+ * Instances are initially constructed by {@link ObjectMapper} and can be
+ * reused.
+ *
+ * @author tatu
+ * @since 1.6
+ */
+
+public class BackportedObjectReader {
+
+    final static Field ROOT_DESERIALIZERS;
+
+    static {
+        Field fl = ReflectionUtils.findField(ObjectMapper.class, "_rootDeserializers");
+        Assert.isNotNull(fl, "Cannot find root deserializers");
+        ROOT_DESERIALIZERS = fl;
+        ReflectionUtils.makeAccessible(fl);
+    }
+
+    /*
+    /**********************************************************
+    /* Immutable configuration from ObjectMapper
+    /**********************************************************
+     */
+
+    /**
+     * Root-level cached deserializers
+     */
+    final protected ConcurrentHashMap<JavaType, JsonDeserializer<Object>> _rootDeserializers;
+
+    /**
+     * General serialization configuration settings
+     */
+    protected final DeserializationConfig _config;
+
+    protected final DeserializerProvider _provider;
+
+    /**
+     * Factory used for constructing {@link JsonGenerator}s
+     */
+    protected final JsonFactory _jsonFactory;
+
+    // Support for polymorphic types:
+    protected TypeResolverBuilder<?> _defaultTyper;
+
+    // Configurable visibility limits
+    protected VisibilityChecker<?> _visibilityChecker;
+
+    /*
+    /**********************************************************
+    /* Configuration that can be changed during building
+    /**********************************************************
+     */
+
+    /**
+     * Declared type of value to instantiate during deserialization.
+     * Defines which deserializer to use; as well as base type of instance
+     * to construct if an updatable value is not configured to be used
+     * (subject to changes by embedded type information, for polymorphic
+     * types). If {@link #_valueToUpdate} is non-null, only used for
+     * locating deserializer.
+     */
+    protected final JavaType _valueType;
+
+    /**
+     * Instance to update with data binding; if any. If null,
+     * a new instance is created, if non-null, properties of
+     * this value object will be updated instead.
+     * Note that value can be of almost any type, except not
+     * {@link org.codehaus.jackson.map.type.ArrayType}; array
+     * types can not be modified because array size is immutable.
+     */
+    protected final Object _valueToUpdate;
+
+
+    public static BackportedObjectReader create(ObjectMapper mapper, Class<?> type) {
+        return new BackportedObjectReader(mapper, TypeFactory.type(type), null);
+    }
+
+    /**
+     * Constructor used by {@link ObjectMapper} for initial instantiation
+     */
+    protected BackportedObjectReader(ObjectMapper mapper, JavaType valueType, Object valueToUpdate) {
+        _rootDeserializers = ReflectionUtils.getField(ROOT_DESERIALIZERS, mapper);
+        _provider = mapper.getDeserializerProvider();
+        _jsonFactory = mapper.getJsonFactory();
+
+        // must make a copy at this point, to prevent further changes from trickling down
+        _config = mapper.copyDeserializationConfig();
+
+        _valueType = valueType;
+        _valueToUpdate = valueToUpdate;
+        if (valueToUpdate != null && valueType.isArrayType()) {
+            throw new IllegalArgumentException("Can not update an array value");
+        }
+    }
+
+
+    public <T> BackportedJacksonMappingIterator<T> readValues(JsonParser jp) throws IOException,
+            JsonProcessingException {
+        DeserializationContext ctxt = _createDeserializationContext(jp, _config);
+        return new BackportedJacksonMappingIterator<T>(_valueType, jp, ctxt, _findRootDeserializer(_config, _valueType));
+    }
+
+
+    /**
+     * Method called to locate deserializer for the passed root-level value.
+     */
+    protected JsonDeserializer<Object> _findRootDeserializer(DeserializationConfig cfg, JavaType valueType)
+            throws JsonMappingException {
+        // First: have we already seen it?
+        JsonDeserializer<Object> deser = _rootDeserializers.get(valueType);
+        if (deser != null) {
+            return deser;
+        }
+
+        // Nope: need to ask provider to resolve it
+        deser = _provider.findTypedValueDeserializer(cfg, valueType);
+        if (deser == null) { // can this happen?
+            throw new JsonMappingException("Can not find a deserializer for type " + valueType);
+        }
+        _rootDeserializers.put(valueType, deser);
+        return deser;
+    }
+
+    protected DeserializationContext _createDeserializationContext(JsonParser jp, DeserializationConfig cfg) {
+        // 04-Jan-2010, tatu: we do actually need the provider too... (for polymorphic deser)
+        return new StdDeserializationContext(cfg, jp, _provider);
+    }
+}