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);  }
@Test void executeSuccessfully() { List<ExprValue> expected = Arrays.asList( tupleValue(of("name", "John", "age", 20)), tupleValue(of("name", "Allen", "age", 30))); FakePhysicalPlan plan = new FakePhysicalPlan(expected.iterator()); when(protector.protect(plan)).thenReturn(plan); ElasticsearchExecutionEngine executor = new ElasticsearchExecutionEngine(client, protector); List<ExprValue> actual = new ArrayList<>(); executor.execute( plan, new ResponseListener<QueryResponse>() { @Override public void onResponse(QueryResponse response) { actual.addAll(response.getResults()); } @Override public void onFailure(Exception e) { fail("Error occurred during execution", e); } }); assertTrue(plan.hasOpen); assertEquals(expected, actual); assertTrue(plan.hasClosed); }
@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(); }