From ab1f17029a02c917aa534a654441e52013737c42 Mon Sep 17 00:00:00 2001
From: Mathias Fussenegger <f.mathias@zignar.net>
Date: Mon, 9 Mar 2015 15:20:00 +0100
Subject: [PATCH] fix exception swallowing in searchResponse callback

---
 .../elasticsearch/QueryThenFetchTask.java     | 21 ++++++++--------
 .../QueryThenFetchIntegrationTest.java        | 24 +++++++++++++++++++
 2 files changed, 35 insertions(+), 10 deletions(-)

diff --git a/sql/src/main/java/io/crate/executor/transport/task/elasticsearch/QueryThenFetchTask.java b/sql/src/main/java/io/crate/executor/transport/task/elasticsearch/QueryThenFetchTask.java
index 8e8669a7385d..4d7894c44812 100644
--- a/sql/src/main/java/io/crate/executor/transport/task/elasticsearch/QueryThenFetchTask.java
+++ b/sql/src/main/java/io/crate/executor/transport/task/elasticsearch/QueryThenFetchTask.java
@@ -44,6 +44,7 @@
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.internal.InternalSearchResponse;
 
+import javax.annotation.Nonnull;
 import java.io.IOException;
 import java.util.*;
 
@@ -102,23 +103,23 @@ public void onSuccess(@Nullable final QueryThenFetchOperation.QueryThenFetchCont
                     Futures.addCallback(context.createSearchResponse(), new FutureCallback<InternalSearchResponse>() {
                         @Override
                         public void onSuccess(@Nullable InternalSearchResponse searchResponse) {
-                            if (pageInfo.isPresent()) {
-                                ObjectArray<Object[]> pageSource = context.toPage(searchResponse.hits().hits(), extractors);
-                                context.cleanAfterFirstPage();
-                                result.set(new QueryThenFetchPageableTaskResult(operation, context, extractors, pageInfo.get(), pageSource, 0L));
-                            } else {
-                                Object[][] rows = context.toRows(searchResponse.hits().hits(), extractors);
-                                try {
+                            try {
+                                if (pageInfo.isPresent()) {
+                                    ObjectArray<Object[]> pageSource = context.toPage(searchResponse.hits().hits(), extractors);
+                                    context.cleanAfterFirstPage();
+                                    result.set(new QueryThenFetchPageableTaskResult(operation, context, extractors, pageInfo.get(), pageSource, 0L));
+                                } else {
+                                    Object[][] rows = context.toRows(searchResponse.hits().hits(), extractors);
                                     context.close();
                                     result.set(new QueryResult(rows));
-                                } catch (IOException e) {
-                                    onFailure(e);
                                 }
+                            } catch (Throwable t) {
+                                onFailure(t);
                             }
                         }
 
                         @Override
-                        public void onFailure(Throwable t) {
+                        public void onFailure(@Nonnull Throwable t) {
                             try {
                                 context.close();
                                 logger.error("error creating a QueryThenFetch response", t);
diff --git a/sql/src/test/java/io/crate/integrationtests/QueryThenFetchIntegrationTest.java b/sql/src/test/java/io/crate/integrationtests/QueryThenFetchIntegrationTest.java
index b55f609ccfee..f0495eb700aa 100644
--- a/sql/src/test/java/io/crate/integrationtests/QueryThenFetchIntegrationTest.java
+++ b/sql/src/test/java/io/crate/integrationtests/QueryThenFetchIntegrationTest.java
@@ -21,14 +21,21 @@
 
 package io.crate.integrationtests;
 
+import io.crate.action.sql.SQLActionException;
 import io.crate.test.integration.CrateIntegrationTest;
+import org.hamcrest.Matchers;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import static org.hamcrest.Matchers.is;
 
 @CrateIntegrationTest.ClusterScope(scope = CrateIntegrationTest.Scope.GLOBAL)
 public class QueryThenFetchIntegrationTest extends SQLTransportIntegrationTest {
 
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
     @Test
     public void testCrateSearchServiceSupportsOrderByOnFunctionWithBooleanReturnType() throws Exception {
         execute("create table t (name string) with (number_of_replicas = 0)");
@@ -40,4 +47,21 @@ public void testCrateSearchServiceSupportsOrderByOnFunctionWithBooleanReturnType
         assertThat(((String) response.rows()[0][0]), is("Trillian"));
         assertThat(((String) response.rows()[1][0]), is("Marvin"));
     }
+
+    @Test
+    public void testThatErrorsInSearchResponseCallbackAreNotSwallowed() throws Exception {
+        expectedException.expect(SQLActionException.class);
+        expectedException.expectMessage(Matchers.anyOf(
+                is("value of argument to extract must be of type long (timestamp)"),
+                is("java.lang.Integer cannot be cast to java.lang.Long")
+        ));
+
+        execute("create table t (d timestamp) clustered into 1 shards with (number_of_replicas = 0)");
+        ensureYellow();
+
+         // timestamp is invalid.. will be an Integer and cause a classCastException in extract because long is required
+        execute("insert into t (d) values (1425910189)");
+        execute("refresh table t");
+        execute("select extract(day from d) from t");
+    }
 }