[jira] [Issue Comment Edited] (CAMEL-4925) ThreadsProcessor configured with ExecutorService with DiscardPolicy or DiscardOldestPolicy leaves inflight exchanges for discarded tasks unprocessed.

["Claus Ibsen (Issue Comment Edited) (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/CAMEL-4925?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13189758#comment-13189758 ] 

Claus Ibsen edited comment on CAMEL-4925 at 1/20/12 12:26 PM:
--------------------------------------------------------------

Yeah we would need to check in ThreadsDefinition if you have configured Discard and DiscardOldest. And then wrap those policies with a custom policy, so we get the callback from the JDK when the task is rejected.

The discard, would possible not be needed, as I would assume the thread pool will reject it asap, when you try to submit it. But the discard oldest, is an existing task from the task queue, so that is a different story.

Then we need to provide this as a callback to the ThreadsProcessor, so it can do custom logic when the RejectedExecutionHandler#rejectedExecution is invoked. We can then from the Runnable parameter cast that to ProcessCall, and then get access to the Exchange. Then we can set on the Exchange a RejectedExecutionException as exception, and invoke its callback. Then Camel will take it from there to remove the exchange from inflight registry and whatnot.

Something along the lines of that. Its a shame the API of the ExecutorService do not have a 2nd optional runnable parameter for rejected execution. Then it would have been easier. As you just submit 2 runnable's. And the JDK will invoke the appropriate.
                
      was (Author: davsclaus):
    Yeah we would need to check in ThreadsDefinition if you have configured Discard and DiscardOldest. And then wrap those policies with a custom policy, so we get the callback from the JDK when the task is rejected.

The discard, would possible not be needed, as I would assume the thread pool will reject it asap, when you try to submit it. But the discard oldest, is an existing task from the task queue, so that is a different story.

Then we need to provide this as a callback to the ThreadsProcessor, so it can do custom logic when the RejectedExecutionHandler#rejectedExecution is invoked. We can then from the Runnable parameter cast that to ProcessCall, and then get access to the Exchange. Then we can set on the Exchange a RejectedExecutionException as exception, and invoke its callback. Then Camel will take it from there to remove the exchange from inflight registry and whatnot.

Something along the lines of that. Its a shame the API of the ExecutorService do not have a runnable for rejected execution. Then it would have been easier.
                  
> ThreadsProcessor configured with ExecutorService with DiscardPolicy or DiscardOldestPolicy leaves inflight exchanges for discarded tasks unprocessed.
> -----------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-4925
>                 URL: https://issues.apache.org/jira/browse/CAMEL-4925
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-core
>    Affects Versions: 2.8.0
>            Reporter: Sergey Zhemzhitsky
>            Priority: Minor
>             Fix For: 2.10.0
>
>         Attachments: CamelRoutingTest.java
>
>
> ThreadsProcessor configured with ExecutorService with DiscardPolicy or DiscardOldestPolicy leaves inflight exchanges for discarded tasks unprocessed.
> Here is the code from ThreadsProcessor. In case of DiscardPolicy or DiscardOldestPolicy executorService will no throw RejectedExecutionException, so exchange remains unprocessed and count of inflight exchanges will not be decremented for such discarded exchanges.
> {code:java|title=ThreadsProcessor#process(Exchange, AsyncCallback)}
> public boolean process(Exchange exchange, AsyncCallback callback) {
>     if (shutdown.get()) {
>         throw new IllegalStateException("ThreadsProcessor is not running.");
>     }
>     ProcessCall call = new ProcessCall(exchange, callback);
>     try {
>         executorService.submit(call);
>         // tell Camel routing engine we continue routing asynchronous
>         return false;
>     } catch (RejectedExecutionException e) {
>         if (isCallerRunsWhenRejected()) {
>             if (shutdown.get()) {
>                 exchange.setException(new RejectedExecutionException());
>             } else {
>                 callback.done(true);
>             }
>         } else {
>             exchange.setException(e);
>         }
>         return true;
>     }
> }
> {code}
> Unit test is attached.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira