Fixed SI-3070 and related issues

RB=820373
BUG=SI-3070
G=si-dev
R=mnchen,kvidhani,axu
A=mnchen,sihde,kvidhani,axu

pegasus-common/src/main/java/com/linkedin/common/callback/CallbackAdapter.java

@@ -17,6 +17,11 @@
 /* $Id$ */
 package com.linkedin.common.callback;
 
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
 /**
  * Adapts the successful result type of a callback to another type.
  *
@@ -25,6 +30,7 @@
  */
 public abstract class CallbackAdapter<OLD, NEW> implements Callback<NEW>
 {
+  private static final Logger LOG = LoggerFactory.getLogger(CallbackAdapter.class);
   private final Callback<OLD> _callback;
 
   protected CallbackAdapter(final Callback<OLD> callback)
@@ -71,6 +77,13 @@ public void onSuccess(final NEW response)
   @Override
   public void onError(final Throwable e)
   {
-    _callback.onError(convertError(e));
+    Throwable newThrowable;
+    try {
+      newThrowable = convertError(e);
+    } catch (RuntimeException ex) {
+      LOG.error("Failed to convert callback error, original exception follows:", e);
+      newThrowable = ex;
+    }
+    _callback.onError(newThrowable);
   }
 }

pegasus-common/src/main/java/com/linkedin/common/callback/FutureCallback.java

@@ -100,12 +100,8 @@ public void onSuccess(final T t)
   @Override
   public void onError(final Throwable e)
   {
-    if (e == null)
-    {
-      throw new NullPointerException();
-    }
-
-    safeSetValue(Result.<T>createError(e));
+    Throwable error = e != null ? e : new NullPointerException("Null error is passed to onError!");
+    safeSetValue(Result.<T>createError(error));
     _doneLatch.countDown();
   }
 

r2-core/src/main/java/com/linkedin/r2/filter/FilterChainIterator.java

@@ -30,6 +30,10 @@
 import java.util.List;
 import java.util.Map;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
 /**
 * @author Chris Pettitt
 * @author Zhenkai Zhu
@@ -38,6 +42,7 @@
 /* package private */ abstract class FilterChainIterator<F, REQ extends Request, RES extends Response>
         implements NextFilter<REQ, RES>
 {
+  private static final Logger LOG = LoggerFactory.getLogger(FilterChainIterator.class);
   private final List<F> _filters;
   private int _cursor;
 
@@ -92,6 +97,9 @@ public void onError(Throwable ex, RequestContext requestContext, Map<String, Str
       {
         onError(e, requestContext, wireAttrs);
       }
+    } else {
+      String lastFilterName = (_filters != null && _filters.size() > 0) ? _filters.get(0).getClass().getName() : "";
+      LOG.error("Uncaught exception from the last response filter in the filter chain: " + lastFilterName, ex);
     }
   }
 

restli-client/src/main/java/com/linkedin/restli/internal/client/BatchEntityResponseDecoder.java

@@ -77,6 +77,11 @@ public Class<?> getEntityClass()
   public BatchKVResponse<K, EntityResponse<V>> wrapResponse(DataMap dataMap, Map<String, String> headers, ProtocolVersion version)
     throws InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException, IOException
   {
+    if (dataMap == null)
+    {
+      return null;
+    }
+
     final DataMap mergedResults = new DataMap();
     final DataMap inputResults = dataMap.containsKey(BatchResponse.RESULTS) ? dataMap.getDataMap(BatchResponse.RESULTS)
                                                                             : new DataMap();

restli-client/src/main/java/com/linkedin/restli/internal/client/BatchUpdateResponseDecoder.java

@@ -71,6 +71,11 @@ public Class<?> getEntityClass()
   public BatchKVResponse<K, UpdateStatus> wrapResponse(DataMap dataMap, Map<String, String> headers, ProtocolVersion version)
     throws InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException, IOException
   {
+    if (dataMap == null)
+    {
+      return null;
+    }
+
     final DataMap mergedResults = new DataMap();
     final DataMap inputResults = dataMap.containsKey(BatchResponse.RESULTS) ? dataMap.getDataMap(BatchResponse.RESULTS)
                                                                             : new DataMap();

restli-client/src/test/java/com/linkedin/restli/internal/client/TestBatchEntityResponseDecoder.java

@@ -122,4 +122,18 @@ private static Object[][] batchEntityResponseDataProvider()
       { _keys, AllProtocolVersions.RESTLI_PROTOCOL_2_0_0.getProtocolVersion() }
     };
   }
+
+  @Test(dataProvider = TestConstants.RESTLI_PROTOCOL_1_2_PREFIX + "batchEntityResponseDataProvider")
+  public void testDecodingWithEmptyDataMap(List<String> keys, ProtocolVersion protocolVersion)
+    throws InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException, IOException
+  {
+    final BatchEntityResponseDecoder<String, TestRecord> decoder =
+      new BatchEntityResponseDecoder<String, TestRecord>(new TypeSpec<TestRecord>(TestRecord.class),
+                                                         new TypeSpec<String>(String.class),
+                                                         Collections.<String, CompoundKey.TypeInfo>emptyMap(),
+                                                         null);
+
+    final BatchKVResponse<String, EntityResponse<TestRecord>> response = decoder.wrapResponse(null, Collections.<String, String>emptyMap(), protocolVersion);
+    Assert.assertNull(response);
+  }
 }

restli-client/src/test/java/com/linkedin/restli/internal/client/TestBatchUpdateResponseDecoder.java

@@ -0,0 +1,63 @@
+/*
+   Copyright (c) 2014 LinkedIn Corp.
+
+   Licensed 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 com.linkedin.restli.internal.client;
+
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.Collections;
+import java.util.Map;
+
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import com.linkedin.restli.client.response.BatchKVResponse;
+import com.linkedin.restli.common.CompoundKey;
+import com.linkedin.restli.common.ErrorResponse;
+import com.linkedin.restli.common.ProtocolVersion;
+import com.linkedin.restli.common.TypeSpec;
+import com.linkedin.restli.common.UpdateStatus;
+import com.linkedin.restli.internal.common.AllProtocolVersions;
+import com.linkedin.restli.internal.common.TestConstants;
+
+
+/**
+ * @author Jun Chen
+ */
+public class TestBatchUpdateResponseDecoder
+{
+  @DataProvider(name = TestConstants.RESTLI_PROTOCOL_1_2_PREFIX + "batchEntityResponseDataProvider")
+  private static Object[][] batchEntityResponseDataProvider()
+  {
+    return new Object[][] {
+      { AllProtocolVersions.RESTLI_PROTOCOL_1_0_0.getProtocolVersion() },
+      { AllProtocolVersions.RESTLI_PROTOCOL_2_0_0.getProtocolVersion() },
+    };
+  }
+
+  @Test(dataProvider = TestConstants.RESTLI_PROTOCOL_1_2_PREFIX + "batchEntityResponseDataProvider")
+  public void testDecodingWithEmptyDataMap(ProtocolVersion protocolVersion)
+    throws InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException, IOException
+  {
+    final BatchUpdateResponseDecoder<String> decoder =
+      new BatchUpdateResponseDecoder<String>(new TypeSpec<String>(String.class), Collections.<String, CompoundKey.TypeInfo>emptyMap(), null);
+
+    final BatchKVResponse<String, UpdateStatus> response = decoder.wrapResponse(null, Collections.<String, String>emptyMap(), protocolVersion);
+    Assert.assertNull(response);
+  }
+}

restli-int-test/src/test/java/com/linkedin/restli/client/TestResponseDecoder.java

@@ -0,0 +1,103 @@
+/*
+   Copyright (c) 2012 LinkedIn Corp.
+
+   Licensed 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 com.linkedin.restli.client;
+
+
+import com.linkedin.common.callback.CallbackAdapter;
+import com.linkedin.restli.examples.RestLiIntegrationTest;
+import com.linkedin.restli.examples.TestConstants;
+import com.linkedin.restli.examples.greetings.api.Message;
+import com.linkedin.restli.examples.greetings.client.StringKeysRequestBuilders;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+
+/**
+ * Tests for client response decoders.
+ *
+ * @author Jun Chen
+ */
+public class TestResponseDecoder extends RestLiIntegrationTest
+{
+  /* header size (in KBs) that likely exceed server request limit and will cause container exception. */
+  private static int SERVER_HEADER_OVERLOAD_SIZE = 50;
+
+  @BeforeClass
+  public void initClass() throws Exception
+  {
+    super.init();
+  }
+
+  @AfterClass
+  public void shutDown() throws Exception
+  {
+    super.shutdown();
+  }
+
+  /**
+   * This test tests 2 things in combo here:
+   * 1) BatchEntityResponseDecoder could be invoked in some cases to try to decode a empty response dataMap when
+   * non-rest.li server error returns, in this test, we simulate that by passing a over-size URL param
+   * {@link com.linkedin.restli.internal.client.ExceptionUtil#exceptionForThrowable(java.lang.Throwable, com.linkedin.restli.internal.client.RestResponseDecoder)}
+   *
+   * 2) CallbackAdapter and its subclasses could have error while its dealing with error itself, this test make sure it
+   * pass the 'new' error to its inner callback's onError method.
+   * {@link CallbackAdapter#onError(java.lang.Throwable)}
+   */
+  @Test(dataProvider = com.linkedin.restli.internal.common.TestConstants.RESTLI_PROTOCOL_1_2_PREFIX + "dataProvider")
+  public void testNonRestliServerErrorHandling(RestliRequestOptions requestOptions) throws Exception
+  {
+    Set<String> keys = new HashSet<String>();
+    keys.add(createDataSize(SERVER_HEADER_OVERLOAD_SIZE));
+    BatchGetEntityRequest<String, Message> req = new StringKeysRequestBuilders(requestOptions).batchGet().ids(keys).build();
+    ResponseFuture batchKVResponseResponseFuture = getClient().sendRequest(req);
+    try {
+      batchKVResponseResponseFuture.getResponse();
+      Assert.fail("Exception should have thrown before this point!");
+    } catch (Throwable e) {
+      Assert.assertTrue(e instanceof RestLiResponseException);
+      Assert.assertEquals(((RestLiResponseException) e).getStatus(), 414);
+    }
+  }
+
+  @DataProvider(name = com.linkedin.restli.internal.common.TestConstants.RESTLI_PROTOCOL_1_2_PREFIX + "dataProvider")
+  private static Object[][] dataProvider()
+  {
+    return new Object[][] {
+      { RestliRequestOptions.DEFAULT_OPTIONS },
+      { TestConstants.FORCE_USE_NEXT_OPTIONS }
+    };
+  }
+
+  /**
+   * Creates a string of size @msgSize in KB.
+   */
+  private static String createDataSize(int msgSize) {
+    msgSize = msgSize / 2 * 1024;
+    StringBuilder sb = new StringBuilder(msgSize);
+    for (int i = 0; i < msgSize; i++)
+        sb.append('a');
+    return sb.toString();
+  }
+}