Thrift: Response might get corrupted with extra data, closes elastic#452.

Author: kimchy

plugins/transport/thrift/src/main/java/org/elasticsearch/thrift/ThriftRestImpl.java

@@ -49,16 +49,20 @@ public class ThriftRestImpl extends AbstractComponent implements Rest.Iface {
             logger.trace("thrift message {}", request);
         }
         final CountDownLatch latch = new CountDownLatch(1);
-        final AtomicReference<RestResponse> ref = new AtomicReference<RestResponse>();
+        final AtomicReference<org.elasticsearch.thrift.RestResponse> ref = new AtomicReference<org.elasticsearch.thrift.RestResponse>();
         restController.dispatchRequest(new ThriftRestRequest(request), new RestChannel() {
             @Override public void sendResponse(RestResponse response) {
-                ref.set(response);
+                try {
+                    ref.set(convert(response));
+                } catch (IOException e) {
+                    // ignore, should not happen...
+                }
                 latch.countDown();
             }
         });
         try {
             latch.await();
-            return convert(ref.get());
+            return ref.get();
         } catch (Exception e) {
             throw new TException("failed to generate response", e);
         }
@@ -67,7 +71,14 @@ public class ThriftRestImpl extends AbstractComponent implements Rest.Iface {
     private org.elasticsearch.thrift.RestResponse convert(RestResponse response) throws IOException {
         org.elasticsearch.thrift.RestResponse tResponse = new org.elasticsearch.thrift.RestResponse(getStatus(response.status()));
         if (response.contentLength() > 0) {
-            tResponse.setBody(ByteBuffer.wrap(response.content(), 0, response.contentLength()));
+            if (response.contentThreadSafe()) {
+                tResponse.setBody(ByteBuffer.wrap(response.content(), 0, response.contentLength()));
+            } else {
+                // argh!, we need to copy it over since we are not on the same thread...
+                byte[] body = new byte[response.contentLength()];
+                System.arraycopy(response.content(), 0, body, 0, response.contentLength());
+                tResponse.setBody(ByteBuffer.wrap(body));
+            }
         }
         return tResponse;
     }