@Override
    public void execute(String mapName, Predicate predicate, Collection<Integer> partitions, Result result) {
        RetryableHazelcastException storedException = null;
        for (Integer partitionId : partitions) {
            try {
                partitionScanRunner.run(mapName, predicate, partitionId, result);
            } catch (RetryableHazelcastException e) {
                // RetryableHazelcastException are stored and re-thrown later. this is to ensure all partitions
                // are touched as when the parallel execution was used.
                // see discussion at https://github.com/hazelcast/hazelcast/pull/5049#discussion_r28773099 for details.
                if (storedException == null) {
                    storedException = e;
                }
            }
        }
        if (storedException != null) {
            throw storedException;
        }
    }
@Test
    public void execute_fail_retryable() {
        PartitionScanRunner runner = mock(PartitionScanRunner.class);
        CallerRunsPartitionScanExecutor executor = new CallerRunsPartitionScanExecutor(runner);
        Predicate<Object, Object> predicate = Predicates.equal("attribute", 1);
        QueryResult queryResult = new QueryResult(IterationType.ENTRY, null, null, Long.MAX_VALUE, false);

        doThrow(new RetryableHazelcastException()).when(runner).run(anyString(), eq(predicate), anyInt(), eq(queryResult));

        List<Integer> list = asList(1, 2, 3);
        assertThatThrownBy(() -> executor.execute("Map", predicate, list, queryResult))
                .isInstanceOf(RetryableHazelcastException.class);
    }