@Test void executeWithFailure() { PhysicalPlan plan = mock(PhysicalPlan.class); RuntimeException expected = new RuntimeException("Execution error"); when(plan.hasNext()).thenThrow(expected); when(protector.protect(plan)).thenReturn(plan); ElasticsearchExecutionEngine executor = new ElasticsearchExecutionEngine(client, protector); AtomicReference<Exception> actual = new AtomicReference<>(); executor.execute( plan, new ResponseListener<QueryResponse>() { @Override public void onResponse(QueryResponse response) { fail("Expected error didn't happen"); } @Override public void onFailure(Exception e) { actual.set(e); } }); assertEquals(expected, actual.get()); verify(plan).close(); }
@Override public void execute(PhysicalPlan physicalPlan, ResponseListener<QueryResponse> listener) { PhysicalPlan plan = executionProtector.protect(physicalPlan); client.schedule( () -> { try { List<ExprValue> result = new ArrayList<>(); plan.open(); while (plan.hasNext()) { result.add(plan.next()); } QueryResponse response = new QueryResponse(physicalPlan.schema(), result); listener.onResponse(response); } catch (Exception e) { listener.onFailure(e); } finally { plan.close(); } }); }
ElasticsearchExecutionEngine implements ExecutionEngine { @Override public void execute(PhysicalPlan physicalPlan, ResponseListener<QueryResponse> listener) { PhysicalPlan plan = executionProtector.protect(physicalPlan); client.schedule( () -> { try { List<ExprValue> result = new ArrayList<>(); plan.open(); while (plan.hasNext()) { result.add(plan.next()); } QueryResponse response = new QueryResponse(physicalPlan.schema(), result); listener.onResponse(response); } catch (Exception e) { listener.onFailure(e); } finally { plan.close(); } }); } }
ElasticsearchExecutionEngine implements ExecutionEngine { @Override public void execute(PhysicalPlan physicalPlan, ResponseListener<QueryResponse> listener) { PhysicalPlan plan = executionProtector.protect(physicalPlan); client.schedule( () -> { try { List<ExprValue> result = new ArrayList<>(); plan.open(); while (plan.hasNext()) { result.add(plan.next()); } QueryResponse response = new QueryResponse(physicalPlan.schema(), result); listener.onResponse(response); } catch (Exception e) { listener.onFailure(e); } finally { plan.close(); } }); }  }
ElasticsearchExecutionEngine implements ExecutionEngine { @Override public void execute(PhysicalPlan physicalPlan, ResponseListener<QueryResponse> listener) { PhysicalPlan plan = executionProtector.protect(physicalPlan); client.schedule( () -> { try { List<ExprValue> result = new ArrayList<>(); plan.open(); while (plan.hasNext()) { result.add(plan.next()); } QueryResponse response = new QueryResponse(physicalPlan.schema(), result); listener.onResponse(response); } catch (Exception e) { listener.onFailure(e); } finally { plan.close(); } }); }  @Override void execute(PhysicalPlan physicalPlan, ResponseListener<QueryResponse> listener); }
ElasticsearchExecutionEngine implements ExecutionEngine { @Override public void execute(PhysicalPlan physicalPlan, ResponseListener<QueryResponse> listener) { PhysicalPlan plan = executionProtector.protect(physicalPlan); client.schedule( () -> { try { List<ExprValue> result = new ArrayList<>(); plan.open(); while (plan.hasNext()) { result.add(plan.next()); } QueryResponse response = new QueryResponse(physicalPlan.schema(), result); listener.onResponse(response); } catch (Exception e) { listener.onFailure(e); } finally { plan.close(); } }); }  @Override void execute(PhysicalPlan physicalPlan, ResponseListener<QueryResponse> listener);  }