'Version 1.0.2 of the Amazon SQS Java Messaging Library'

Author: kuba-aws

NOTICE.txt

@@ -1,5 +1,5 @@
 Amazon SQS Java Messaging Library
-Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 
 This product includes software developed by
 Amazon Technologies, Inc (http://www.amazon.com/).

README.md

@@ -16,7 +16,7 @@ for communicating with Amazon Simple Queue Service. This project builds on top o
   <dependency>
     <groupId>com.amazonaws</groupId>
     <artifactId>amazon-sqs-java-messaging-lib</artifactId>
-    <version>1.0.1</version>
+    <version>1.0.2</version>
     <type>jar</type>
   </dependency>
 ```

pom.xml

@@ -6,7 +6,7 @@
 
   <groupId>com.amazonaws</groupId>
   <artifactId>amazon-sqs-java-messaging-lib</artifactId>
-  <version>1.0.1</version>
+  <version>1.0.2</version>
   <packaging>jar</packaging>
   <name>Amazon SQS Java Messaging Library</name>
   <description>The Amazon SQS Java Messaging Library holds the Java Message Service compatible classes, that are used
@@ -38,7 +38,7 @@
   </developers>
 
   <properties>
-    <aws-java-sdk.version>1.11.18</aws-java-sdk.version>
+    <aws-java-sdk.version>1.11.106</aws-java-sdk.version>
   </properties>
 
   <dependencies>

src/main/java/com/amazon/sqs/javamessaging/AmazonSQSMessagingClientWrapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.

src/main/java/com/amazon/sqs/javamessaging/PrefetchManager.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.

src/main/java/com/amazon/sqs/javamessaging/SQSConnection.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.

src/main/java/com/amazon/sqs/javamessaging/SQSConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.
@@ -131,8 +131,7 @@ public Builder(String region) {
 
         public Builder() {
             clientConfiguration = new ClientConfiguration();
-            clientConfiguration.setUserAgent(
-                    clientConfiguration.getUserAgent() + SQSMessagingClientConstants.APPENDED_USER_AGENT_HEADER_VERSION );
+            clientConfiguration.setUserAgentPrefix(clientConfiguration.getUserAgentPrefix() + SQSMessagingClientConstants.APPENDED_USER_AGENT_HEADER_VERSION );
 
             // Set default numberOfMessagesToPrefetch to MIN_BATCH.
             this.numberOfMessagesToPrefetch = SQSMessagingClientConstants.MIN_BATCH;
@@ -220,11 +219,11 @@ public ClientConfiguration getClientConfiguration() {
 
         public void setClientConfiguration(ClientConfiguration clientConfig) {
             clientConfiguration = new ClientConfiguration( clientConfig );
-            if( clientConfig.getUserAgent() == null || clientConfig.getUserAgent().isEmpty() ) {
-                clientConfig.setUserAgent( ClientConfiguration.DEFAULT_USER_AGENT );
+            if( clientConfig.getUserAgentPrefix() == null || clientConfig.getUserAgentPrefix().isEmpty() ) {
+                clientConfig.setUserAgentPrefix( ClientConfiguration.DEFAULT_USER_AGENT );
             }
-            clientConfiguration.setUserAgent(
-                    clientConfig.getUserAgent() + SQSMessagingClientConstants.APPENDED_USER_AGENT_HEADER_VERSION );
+            clientConfiguration.setUserAgentPrefix(
+                    clientConfig.getUserAgentPrefix() + SQSMessagingClientConstants.APPENDED_USER_AGENT_HEADER_VERSION );
         }
 
         public int getNumberOfMessagesToPrefetch() {

src/main/java/com/amazon/sqs/javamessaging/SQSConnectionMetaData.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.

src/main/java/com/amazon/sqs/javamessaging/SQSMessageConsumer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.
@@ -15,6 +15,7 @@
 package com.amazon.sqs.javamessaging;
 
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
@@ -31,7 +32,6 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-
 import com.amazon.sqs.javamessaging.acknowledge.Acknowledger;
 import com.amazon.sqs.javamessaging.acknowledge.NegativeAcknowledger;
 import com.amazon.sqs.javamessaging.acknowledge.SQSMessageIdentifier;
@@ -57,11 +57,8 @@ public class SQSMessageConsumer implements MessageConsumer, QueueReceiver {
     protected volatile boolean closed = false;
 
     private final SQSQueueDestination sqsDestination;
-    private final Acknowledger acknowledger;
     private final SQSSession parentSQSSession;
 
-    private final NegativeAcknowledger negativeAcknowledger;
-
     private final SQSSessionCallbackScheduler sqsSessionRunnable; 
 
     /**
@@ -92,11 +89,9 @@ public class SQSMessageConsumer implements MessageConsumer, QueueReceiver {
                        SQSMessageConsumerPrefetch sqsMessageConsumerPrefetch) {
         this.parentSQSSession = parentSQSSession;
         this.sqsDestination = destination;
-        this.acknowledger = acknowledger;
         this.sqsSessionRunnable = sqsSessionRunnable;
         this.sqsMessageConsumerPrefetch = sqsMessageConsumerPrefetch;
         this.sqsMessageConsumerPrefetch.setMessageConsumer(this);
-        this.negativeAcknowledger = negativeAcknowledger;
 
         prefetchExecutor = Executors.newSingleThreadExecutor(threadFactory);
         prefetchExecutor.execute(sqsMessageConsumerPrefetch);
@@ -249,21 +244,6 @@ void doClose() {
         closed = true;
     }
 
-
-    /**
-     * Set the message visibility as zero for the list of messages which are not
-     * acknowledged and delete them from the list of unacknowledged messages.
-     */
-    void recover() throws JMSException {
-        List<SQSMessageIdentifier> unAckedMessages = acknowledger.getUnAckMessages();
-        if (!unAckedMessages.isEmpty()) {
-            negativeAcknowledger.bulkAction(unAckedMessages, unAckedMessages.size());
-            acknowledger.forgetUnAckMessages();
-        }
-
-        
-    }
-    
     boolean isClosed() {
         return closed;
     }
@@ -293,4 +273,8 @@ private void checkClosed() throws IllegalStateException {
             throw new IllegalStateException("Consumer is closed");
         }
     }
+
+    List<SQSMessageIdentifier> purgePrefetchedMessagesWithGroups(Set<String> affectedGroups) throws JMSException {
+        return sqsMessageConsumerPrefetch.purgePrefetchedMessagesWithGroups(affectedGroups);
+    }
 }

src/main/java/com/amazon/sqs/javamessaging/SQSMessageConsumerPrefetch.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.
@@ -16,17 +16,20 @@
 
 import java.util.ArrayDeque;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Set;
+import java.util.UUID;
 
 import javax.jms.JMSException;
 import javax.jms.MessageListener;
 
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import com.amazon.sqs.javamessaging.acknowledge.Acknowledger;
 import com.amazon.sqs.javamessaging.acknowledge.NegativeAcknowledger;
+import com.amazon.sqs.javamessaging.acknowledge.SQSMessageIdentifier;
 import com.amazon.sqs.javamessaging.message.SQSBytesMessage;
 import com.amazon.sqs.javamessaging.message.SQSMessage;
 import com.amazon.sqs.javamessaging.message.SQSObjectMessage;
@@ -156,9 +159,10 @@ protected void setMessageListener(MessageListener messageListener) {
             if (!running || isClosed()) {
                 return;
             }
-            while (!messageQueue.isEmpty()) {
-                sqsSessionRunnable.scheduleCallBack(messageListener, messageQueue.pollFirst());
-            }
+            
+            List<MessageManager> allPrefetchedMessages = new ArrayList<MessageManager>(messageQueue);
+            sqsSessionRunnable.scheduleCallBacks(messageListener, allPrefetchedMessages);
+            messageQueue.clear();
         }
     }
 
@@ -203,7 +207,7 @@ public void run() {
             } catch (Throwable e) {
                 LOG.error("Unexpected exception when prefetch messages:", e);
                 nackQueueMessages = true;
-                throw e;
+                throw new RuntimeException(e);
             } finally {
                 if (isClosed() || nackQueueMessages) {
                     nackQueueMessages();
@@ -225,6 +229,11 @@ protected List<Message> getMessages(int prefetchBatchSize) throws InterruptedExc
                                                               .withAttributeNames(ALL)
                                                               .withMessageAttributeNames(ALL)
                                                               .withWaitTimeSeconds(WAIT_TIME_SECONDS);
+        //if the receive request is for FIFO queue, provide a unique receive request attempt it, so that
+        //failed calls retried by SDK will claim the same messages
+        if (sqsDestination.isFifo()) {
+            receiveMessageRequest.withReceiveRequestAttemptId(UUID.randomUUID().toString());
+        }
         List<Message> messages = null;
         try {
             ReceiveMessageResult receivedMessageResult = amazonSQSClient.receiveMessage(receiveMessageRequest);
@@ -250,27 +259,26 @@ protected List<Message> getMessages(int prefetchBatchSize) throws InterruptedExc
      */
     protected void processReceivedMessages(List<Message> messages) {
         List<String> nackMessages = new ArrayList<String>();
+        List<MessageManager> messageManagers = new ArrayList<MessageManager>();
         for (Message message : messages) {
             try {
                 javax.jms.Message jmsMessage = convertToJMSMessage(message);
-
-                if (messageListener != null) {
-                    sqsSessionRunnable.scheduleCallBack(messageListener, new MessageManager(this, jmsMessage));
-                    synchronized (stateLock) {
-                        messagesPrefetched++;                        
-                        notifyStateChange();
-                    }
-                } else {
-                    synchronized (stateLock) {
-                        messageQueue.addLast(new MessageManager(this, jmsMessage));
-                        messagesPrefetched++;
-                        notifyStateChange();
-                    }
-                }
+                messageManagers.add(new MessageManager(this, jmsMessage));
             } catch (JMSException e) {
                 nackMessages.add(message.getReceiptHandle());
             }
         }
+        
+        synchronized (stateLock) {
+            if (messageListener != null) {
+                sqsSessionRunnable.scheduleCallBacks(messageListener, messageManagers);
+            } else {
+                messageQueue.addAll(messageManagers);
+            }
+            
+            messagesPrefetched += messageManagers.size();
+            notifyStateChange();
+        }
 
         // Nack any messages that cannot be serialized to JMSMessage.
         try {
@@ -516,4 +524,31 @@ protected void sleep(long sleepTimeMillis) throws InterruptedException {
     protected boolean isClosed() {
         return closed;
     }
+
+    List<SQSMessageIdentifier> purgePrefetchedMessagesWithGroups(Set<String> affectedGroups) throws JMSException {
+        List<SQSMessageIdentifier> purgedMessages = new ArrayList<SQSMessageIdentifier>();
+        synchronized (stateLock) {
+            //let's walk over the prefetched messages
+            Iterator<MessageManager> managerIterator = messageQueue.iterator();
+            while (managerIterator.hasNext()) {
+                MessageManager messageManager = managerIterator.next();
+                SQSMessage prefetchedMessage = (SQSMessage)messageManager.getMessage();
+                SQSMessageIdentifier messageIdentifier = SQSMessageIdentifier.fromSQSMessage(prefetchedMessage);
+
+                //is the prefetch entry for one of the affected group ids?
+                if (affectedGroups.contains(messageIdentifier.getGroupId())) {
+                    //we will purge this prefetched message
+                    purgedMessages.add(messageIdentifier);
+                    //remove from prefetch queue
+                    managerIterator.remove();
+                    //we are done with it and can prefetch more messages
+                    this.messagesPrefetched--;
+                }
+            }
+
+            notifyStateChange();
+        }
+        
+        return purgedMessages;
+    }
 }