[camel] branch main updated: CAMEL-17156: camel-aws2 - do not interrupt extension (#6362)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 87f1163  CAMEL-17156: camel-aws2 - do not interrupt extension (#6362)
87f1163 is described below

commit 87f11631fa37522f6f5de158d81cc8c8c5cb7234
Author: Simon Rasmussen <sr...@viabill.com>
AuthorDate: Tue Nov 2 12:04:48 2021 +0100

    CAMEL-17156: camel-aws2 - do not interrupt extension (#6362)
    
    * CAMEL-17156: camel-aws2 - do not interrupt extension
    
    * CAMEL-17156: camel-aws2 - Catch and swallow SqsException when failing to extend visibility
---
 .../apache/camel/component/aws2/sqs/Sqs2Consumer.java | 19 +++++++++++++++----
 1 file changed, 15 insertions(+), 4 deletions(-)

diff --git a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
index 3d24fa0..a52d77e5 100644
--- a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
+++ b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
@@ -53,6 +53,7 @@ import software.amazon.awssdk.services.sqs.model.QueueDoesNotExistException;
 import software.amazon.awssdk.services.sqs.model.ReceiptHandleIsInvalidException;
 import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
 import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
+import software.amazon.awssdk.services.sqs.model.SqsException;
 
 /**
  * A Consumer of messages from the Amazon Web Service Simple Queue Service <a href="http://aws.amazon.com/sqs/">AWS
@@ -195,7 +196,7 @@ public class Sqs2Consumer extends ScheduledBatchPollingConsumer {
                         // cancel task as we are done
                         LOG.trace("Processing done so cancelling TimeoutExtender task for exchangeId: {}",
                                 exchange.getExchangeId());
-                        scheduledFuture.cancel(true);
+                        scheduledFuture.cancel(false);
                     }
                 });
             }
@@ -384,12 +385,22 @@ public class Sqs2Consumer extends ScheduledBatchPollingConsumer {
                 // Ignore.
             } catch (MessageNotInflightException e) {
                 // Ignore.
+            } catch (SqsException e) {
+                if (e.getMessage().contains("Message does not exist or is not available for visibility timeout change")) {
+                    // Ignore.
+                } else {
+                    logException(e);
+                }
             } catch (Exception e) {
-                LOG.warn("Extending visibility window failed for exchange " + exchange
-                         + ". Will not attempt to extend visibility further. This exception will be ignored.",
-                        e);
+                logException(e);
             }
         }
+
+        private void logException(Exception e) {
+            LOG.warn("Extending visibility window failed for exchange " + exchange
+                     + ". Will not attempt to extend visibility further. This exception will be ignored.",
+                    e);
+        }
     }
 
 }