Unable to catch thrown exception from async method in Spring

Question

I am unable to catch thrown exceptions from an async method in Spring. I have written an uncaught exception handler to catch but was unsuccessful. The application will enable to start any number of forever running asynchronous jobs. I think my async method needs to return Future so that I can store it in hashmap and check its status or stop the job. I also can get all running jobs by storing it. I think I can't use get method of future because if the input is correct it blocks and my job will be forever running. I need to send status as started if the input is fine. Whenever an exception occurs in the Async method it is thrown but I am unable to catch it. How can I do that? Here is my complete code.

Application.java

@SpringBootApplication
@EnableAsync
public class Application  {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

AsyncConfig.java

@EnableAsync
@Configuration
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("MyExecutor-");
        executor.initialize();
        return executor;
    }


    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncExceptionHandler();
    }
}

AsyncExceptionHandler.java

public class AsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
    @Override
    public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {

        System.out.println("Exception Cause - " + throwable.getMessage());
        System.out.println("Method name - " + method.getName());
        for (Object param : obj) {
            System.out.println("Parameter value - " + param);
        }
    }
}

createBucket.java

@Service
public class createBucket {
    @Async
    public Future<String> start(String config){
        try {
            JSONObject map = new JSONObject(config);
            Jedis jedis = new Jedis(map.getString("jedisip"));
            jedis.auth(map.getString("password"));
            // code to make a kafka consumer subscribe to a topic given in config input
            while(true) {
                //forever running code which polls using a kafka consumer
            }
        }
        catch(JedisException j) {
            throw new JedisException("Some msg");
        }
    }
}

Endpoint.java

@Controller
public class Endpoint {
    @Autowired
    private createBucket service;

    private Future<String> out;
    private HashMap<String, Future<String>> maps = new HashMap<>();

    @PostMapping(value = "/start", consumes = "application/json", produces = "application/json")
    public ResponseEntity<String> starttask(@RequestBody String conf) {
        try {
        out = service.start(conf);
        maps.put(conf, out);
    }
    catch (Exception e) {
        return new ResponseEntity<>("exception", HttpStatus.BAD_REQUEST);
    }
        return new ResponseEntity<>("{\"started\":\"true\"}", HttpStatus.CREATED);
    }
}

Answer 1

As stated in official doc, AsyncUncaughtExceptionHandler is used for void return value. https://docs.spring.io/spring/docs/5.1.10.RELEASE/spring-framework-reference/integration.html#spring-integration

In your scenario, I recommend using CompletableFuture and DeferredResult:

    @Async
    public CompletableFuture<String> start(String config) {
        CompletableFuture completableFuture = new CompletableFuture();
        try {
            JSONObject map = new JSONObject(config);
            Jedis jedis = new Jedis(map.getString("jedisip"));
            jedis.auth(map.getString("password"));
            completableFuture.complete("started!");
        }
        catch(JedisException j) {
            completableFuture.completeExceptionally(j);
        }

        return completableFuture;
    }

    @PostMapping(value = "/start", consumes = "application/json", produces = "application/json")
    public DeferredResult<ResponseEntity> starttask(@RequestBody String conf) {

        CompletableFuture<String> start = service.start(conf);

        DeferredResult<ResponseEntity> deferredResult = new DeferredResult<>();

        start.whenComplete((res, ex) -> {
            if (ex == null) {
                ResponseEntity<String> successEntity = new ResponseEntity<>("{\"started\":\"true\"}", HttpStatus.CREATED);\
                deferredResult.setResult(successEntity);
            } else {
                // handle ex here!
                ResponseEntity<String> exEntity = new ResponseEntity<>("exception", HttpStatus.BAD_REQUEST);
                deferredResult.setResult(exEntity);
            }
        });

        return deferredResult;

    }

There is another serious problem. The following code is not thread safe.

private Future<String> out;
private HashMap<String, Future<String>> maps = new HashMap<>();