@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();
}