[jira] [Commented] (CAMEL-12603) Thread stuck in re-delivery loop after interrupting it
[
https://issues.apache.org/jira/browse/CAMEL-12603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16532414#comment-16532414
]
ASF GitHub Bot commented on CAMEL-12603:
oscerd commented on issue #2396: CAMEL-12603 - Interrupt fix for messages stuck
in a re-delivery loop
URL: https://github.com/apache/camel/pull/2396#issuecomment-402401097
Merged on master, 2.22.x and 2.21.x
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Thread stuck in re-delivery loop after interrupting it
> --
>
> Key: CAMEL-12603
> URL: https://issues.apache.org/jira/browse/CAMEL-12603
> Project: Camel
> Issue Type: Bug
> Components: camel-core
>Reporter: Nick Horne
>Priority: Major
> Fix For: 2.21.2, 2.22.1, 2.23.0
>
>
> I have experienced an issue where we could not cancel a message stuck in a
> re-delivery cycle. I was using Jolokia and calling the interrupt method on
> the DefaultAsyncProcessorAwaitManager for the blocked exchange and I had
> expected the re-delivery cycle to stop.
> This does not happen, and the blocked message continues to get executed and
> re-delivered. The mapping does get removed from the in-flight messages
> though. I can see also that the RejectedExecutionException set by the
> interrupt is also overwritten by the exception thrown by our failing bean. I
> think the problem here is that there are no checks for this
> RejectedExecutionException during the re-delivery cycle.
> It seems like the following part of the RedeliveryErrorHandler::call should
> pick up the fact that the exchange has been interrupted:
> {code:java}
> // only process if the exchange hasn't failed
> // and it has not been handled by the error processor
> if (isDone(exchange)) {
> callback.done(false);
> return;
> }{code}
> This is an issue if you have configured a long re-delivery cycle and you have
> a message retrying that you know will never succeed.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (CAMEL-12603) Thread stuck in re-delivery loop after interrupting it
[
https://issues.apache.org/jira/browse/CAMEL-12603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16532415#comment-16532415
]
ASF GitHub Bot commented on CAMEL-12603:
oscerd closed pull request #2396: CAMEL-12603 - Interrupt fix for messages
stuck in a re-delivery loop
URL: https://github.com/apache/camel/pull/2396
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
b/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
index c8dafaa19d7..4a9ef4ddf53 100644
---
a/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
+++
b/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
@@ -161,6 +161,7 @@ public void interrupt(Exchange exchange) {
interruptedCounter.incrementAndGet();
}
exchange.setException(new
RejectedExecutionException("Interrupted while waiting for asynchronous callback
for exchangeId: " + exchange.getExchangeId()));
+exchange.setProperty(Exchange.INTERRUPTED, Boolean.TRUE);
entry.getLatch().countDown();
}
}
diff --git
a/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptWithRedeliveryTest.java
b/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptWithRedeliveryTest.java
new file mode 100644
index 000..c400d4d9756
--- /dev/null
+++
b/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptWithRedeliveryTest.java
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor.async;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.AsyncProcessorAwaitManager;
+import org.apache.camel.util.jndi.JndiContext;
+
+import javax.naming.Context;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Mockito.atMost;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+public class AsyncProcessorAwaitManagerInterruptWithRedeliveryTest extends
ContextTestSupport {
+private CountDownLatch latch;
+private MyBean bean;
+
+@Override
+protected void setUp() throws Exception {
+latch = new CountDownLatch(2);
+bean = spy(new MyBean(latch));
+super.setUp();
+}
+
+public void testAsyncAwaitInterrupt() throws Exception {
+
context.getAsyncProcessorAwaitManager().getStatistics().setStatisticsEnabled(true);
+
+assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+
+getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
+getMockEndpoint("mock:result").expectedMessageCount(0);
+getMockEndpoint("mock:error").expectedMessageCount(0);
+
+createThreadToInterrupt();
+try {
+template.sendBody("direct:start", "Hello Camel");
+fail("Should throw exception");
+} catch (CamelExecutionException e) {
+RejectedExecutionException cause =
assertIsInstanceOf(RejectedExecutionException.class, e.getCause());
+assertTrue(cause.getMessage().startsWith("Interrupted while
waiting for asynchronous callback"));
+}
+
+assertMockEndpointsSatisfied();
+
+// Check we have not reached the full 5 re-deliveries
+verify(bean, atMost(4)).callMe();
+
+assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+assertEquals(1,
context.getAsyncProcessorAwaitManager().getStatisti
[jira] [Commented] (CAMEL-12603) Thread stuck in re-delivery loop after interrupting it
[
https://issues.apache.org/jira/browse/CAMEL-12603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16532416#comment-16532416
]
ASF GitHub Bot commented on CAMEL-12603:
Github user oscerd closed the pull request at:
https://github.com/apache/camel/pull/2396
> Thread stuck in re-delivery loop after interrupting it
> --
>
> Key: CAMEL-12603
> URL: https://issues.apache.org/jira/browse/CAMEL-12603
> Project: Camel
> Issue Type: Bug
> Components: camel-core
>Reporter: Nick Horne
>Priority: Major
> Fix For: 2.21.2, 2.22.1, 2.23.0
>
>
> I have experienced an issue where we could not cancel a message stuck in a
> re-delivery cycle. I was using Jolokia and calling the interrupt method on
> the DefaultAsyncProcessorAwaitManager for the blocked exchange and I had
> expected the re-delivery cycle to stop.
> This does not happen, and the blocked message continues to get executed and
> re-delivered. The mapping does get removed from the in-flight messages
> though. I can see also that the RejectedExecutionException set by the
> interrupt is also overwritten by the exception thrown by our failing bean. I
> think the problem here is that there are no checks for this
> RejectedExecutionException during the re-delivery cycle.
> It seems like the following part of the RedeliveryErrorHandler::call should
> pick up the fact that the exchange has been interrupted:
> {code:java}
> // only process if the exchange hasn't failed
> // and it has not been handled by the error processor
> if (isDone(exchange)) {
> callback.done(false);
> return;
> }{code}
> This is an issue if you have configured a long re-delivery cycle and you have
> a message retrying that you know will never succeed.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (CAMEL-12603) Thread stuck in re-delivery loop after interrupting it
[
https://issues.apache.org/jira/browse/CAMEL-12603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16532322#comment-16532322
]
ASF GitHub Bot commented on CAMEL-12603:
oscerd commented on issue #2396: CAMEL-12603 - Interrupt fix for messages stuck
in a re-delivery loop
URL: https://github.com/apache/camel/pull/2396#issuecomment-402377752
I'm preparing to merge this one.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Thread stuck in re-delivery loop after interrupting it
> --
>
> Key: CAMEL-12603
> URL: https://issues.apache.org/jira/browse/CAMEL-12603
> Project: Camel
> Issue Type: Bug
> Components: camel-core
>Reporter: Nick Horne
>Priority: Major
> Fix For: 2.21.2, 2.22.1, 2.23.0
>
>
> I have experienced an issue where we could not cancel a message stuck in a
> re-delivery cycle. I was using Jolokia and calling the interrupt method on
> the DefaultAsyncProcessorAwaitManager for the blocked exchange and I had
> expected the re-delivery cycle to stop.
> This does not happen, and the blocked message continues to get executed and
> re-delivered. The mapping does get removed from the in-flight messages
> though. I can see also that the RejectedExecutionException set by the
> interrupt is also overwritten by the exception thrown by our failing bean. I
> think the problem here is that there are no checks for this
> RejectedExecutionException during the re-delivery cycle.
> It seems like the following part of the RedeliveryErrorHandler::call should
> pick up the fact that the exchange has been interrupted:
> {code:java}
> // only process if the exchange hasn't failed
> // and it has not been handled by the error processor
> if (isDone(exchange)) {
> callback.done(false);
> return;
> }{code}
> This is an issue if you have configured a long re-delivery cycle and you have
> a message retrying that you know will never succeed.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (CAMEL-12603) Thread stuck in re-delivery loop after interrupting it
[
https://issues.apache.org/jira/browse/CAMEL-12603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16531374#comment-16531374
]
ASF GitHub Bot commented on CAMEL-12603:
NickUK commented on a change in pull request #2396: CAMEL-12603 - Interrupt fix
for messages stuck in a re-delivery loop
URL: https://github.com/apache/camel/pull/2396#discussion_r199804106
##
File path:
camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptWithRedeliveryTest.java
##
@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor.async;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.AsyncProcessorAwaitManager;
+import org.apache.camel.util.jndi.JndiContext;
+
+import javax.naming.Context;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Mockito.atMost;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+public class AsyncProcessorAwaitManagerInterruptWithRedeliveryTest extends
ContextTestSupport {
+private CountDownLatch latch;
+private MyBean bean;
+
+@Override
+protected void setUp() throws Exception {
+latch = new CountDownLatch(2);
+bean = spy(new MyBean(latch));
+super.setUp();
+}
+
+public void testAsyncAwaitInterrupt() throws Exception {
+
context.getAsyncProcessorAwaitManager().getStatistics().setStatisticsEnabled(true);
+
+assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+
+getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
+getMockEndpoint("mock:result").expectedMessageCount(0);
+getMockEndpoint("mock:error").expectedMessageCount(0);
+
+createThreadToInterrupt();
+try {
+template.sendBody("direct:start", "Hello Camel");
+fail("Should throw exception");
+} catch (CamelExecutionException e) {
+RejectedExecutionException cause =
assertIsInstanceOf(RejectedExecutionException.class, e.getCause());
+assertTrue(cause.getMessage().startsWith("Interrupted while
waiting for asynchronous callback"));
+}
+
+assertMockEndpointsSatisfied();
+
+// Check we have not reached the full 5 re-deliveries
+verify(bean, atMost(4)).callMe();
+
+assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+assertEquals(1,
context.getAsyncProcessorAwaitManager().getStatistics().getThreadsBlocked());
+assertEquals(1,
context.getAsyncProcessorAwaitManager().getStatistics().getThreadsInterrupted());
+}
+
+private void createThreadToInterrupt() {
+new Thread(() -> {
+// Allow some time for camel exchange to enter the re-deliveries
+try {
+latch.await(1, TimeUnit.SECONDS);
+} catch (InterruptedException e) {
+e.printStackTrace();
+}
+
+// Get our blocked thread
+int size = context.getAsyncProcessorAwaitManager().size();
+System.out.println("In-flight messages: " + size);
Review comment:
I have removed these
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Thread stuck in re-delivery loop after interrupting it
> --
>
> Key: CAMEL-12603
> URL: https://issues.apache.org/jira/browse/CAMEL-12603
> Project: Camel
> Issue Type: Bug
> Components: camel-core
>Reporter: Nick Horne
>Priority: Major
>
[jira] [Commented] (CAMEL-12603) Thread stuck in re-delivery loop after interrupting it
[
https://issues.apache.org/jira/browse/CAMEL-12603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16531273#comment-16531273
]
ASF GitHub Bot commented on CAMEL-12603:
dmvolod commented on a change in pull request #2396: CAMEL-12603 - Interrupt
fix for messages stuck in a re-delivery loop
URL: https://github.com/apache/camel/pull/2396#discussion_r199783067
##
File path:
camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptWithRedeliveryTest.java
##
@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor.async;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.AsyncProcessorAwaitManager;
+import org.apache.camel.util.jndi.JndiContext;
+
+import javax.naming.Context;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Mockito.atMost;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+public class AsyncProcessorAwaitManagerInterruptWithRedeliveryTest extends
ContextTestSupport {
+private CountDownLatch latch;
+private MyBean bean;
+
+@Override
+protected void setUp() throws Exception {
+latch = new CountDownLatch(2);
+bean = spy(new MyBean(latch));
+super.setUp();
+}
+
+public void testAsyncAwaitInterrupt() throws Exception {
+
context.getAsyncProcessorAwaitManager().getStatistics().setStatisticsEnabled(true);
+
+assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+
+getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
+getMockEndpoint("mock:result").expectedMessageCount(0);
+getMockEndpoint("mock:error").expectedMessageCount(0);
+
+createThreadToInterrupt();
+try {
+template.sendBody("direct:start", "Hello Camel");
+fail("Should throw exception");
+} catch (CamelExecutionException e) {
+RejectedExecutionException cause =
assertIsInstanceOf(RejectedExecutionException.class, e.getCause());
+assertTrue(cause.getMessage().startsWith("Interrupted while
waiting for asynchronous callback"));
+}
+
+assertMockEndpointsSatisfied();
+
+// Check we have not reached the full 5 re-deliveries
+verify(bean, atMost(4)).callMe();
+
+assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+assertEquals(1,
context.getAsyncProcessorAwaitManager().getStatistics().getThreadsBlocked());
+assertEquals(1,
context.getAsyncProcessorAwaitManager().getStatistics().getThreadsInterrupted());
+}
+
+private void createThreadToInterrupt() {
+new Thread(() -> {
+// Allow some time for camel exchange to enter the re-deliveries
+try {
+latch.await(1, TimeUnit.SECONDS);
+} catch (InterruptedException e) {
+e.printStackTrace();
+}
+
+// Get our blocked thread
+int size = context.getAsyncProcessorAwaitManager().size();
+System.out.println("In-flight messages: " + size);
Review comment:
Please avoid using System.out.println if it's not required for use-case. Use
logs for it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Thread stuck in re-delivery loop after interrupting it
> --
>
> Key: CAMEL-12603
> URL: https://issues.apache.org/jira/browse/CAMEL-12603
> Project: Camel
> Issue Type: Bug
> Components: camel-core
>
[jira] [Commented] (CAMEL-12603) Thread stuck in re-delivery loop after interrupting it
[
https://issues.apache.org/jira/browse/CAMEL-12603?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16531262#comment-16531262
]
ASF GitHub Bot commented on CAMEL-12603:
davsclaus commented on issue #2396: CAMEL-12603 - Interrupt fix for messages
stuck in a re-delivery loop
URL: https://github.com/apache/camel/pull/2396#issuecomment-402132134
Lets wait to merge this until we get the 2.22.x branch
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Thread stuck in re-delivery loop after interrupting it
> --
>
> Key: CAMEL-12603
> URL: https://issues.apache.org/jira/browse/CAMEL-12603
> Project: Camel
> Issue Type: Bug
> Components: camel-core
>Reporter: Nick Horne
>Priority: Major
> Fix For: 2.21.2, 2.22.1, 2.23.0
>
>
> I have experienced an issue where we could not cancel a message stuck in a
> re-delivery cycle. I was using Jolokia and calling the interrupt method on
> the DefaultAsyncProcessorAwaitManager for the blocked exchange and I had
> expected the re-delivery cycle to stop.
> This does not happen, and the blocked message continues to get executed and
> re-delivered. The mapping does get removed from the in-flight messages
> though. I can see also that the RejectedExecutionException set by the
> interrupt is also overwritten by the exception thrown by our failing bean. I
> think the problem here is that there are no checks for this
> RejectedExecutionException during the re-delivery cycle.
> It seems like the following part of the RedeliveryErrorHandler::call should
> pick up the fact that the exchange has been interrupted:
> {code:java}
> // only process if the exchange hasn't failed
> // and it has not been handled by the error processor
> if (isDone(exchange)) {
> callback.done(false);
> return;
> }{code}
> This is an issue if you have configured a long re-delivery cycle and you have
> a message retrying that you know will never succeed.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
