Describe the bug The polling of SQS messages for various queues is working fine but the SQS stopped polling twice and all of the SQS messages started to add up in queue. I restarted the server and the queues started to be polled and process instantly. After a month same thing happened and SQS stopped polling and messages started to add up in queues, restarted the server and SQS began to poll and process queues. I've been using several SQS for a long time but this has happened twice in a span of 2 months on production server. Also no error was triggered on events using 'error' and 'processing_error'.

Version of sqs-consumer is "^5.4.0". Version of aws-sdk": "^2.585.0

I've seen a similar issue being reported previously but that was closed with a fix. https://github.com/bbc/sqs-consumer/issues/130 Any idea what could possibly be the reason ?

To Reproduce

Expected behaviour Do not expect SQS to stop polling. Also why did the server restart solved the issue and SQS started to poll again which was in a halt state.

screenshots

Additional context I'm using node.js (v10.19.0) deployed on docker.

Hi,

We're using sqs-consumer to poll our queue, but something strange happens; at times of activity, after a few minutes, the consumer just stops polling for new messages. If we restart our server, then it picks up the messages again.

Here at 22:30 we restarted the server, and it started trying to receive again. Between 19:15 and 22:30 it did not pick any message from the queue. We're only specifying queueUrl - all other options are default. (Also there was no network outage or special event going on., this happens frequently)

Our queue is configured like so :

We're logging errors :

 this._consumer.on("error", e => {
            logger.error(`Error with queue`, { queue: this._queueName, error: e.message });
 });

But this never triggers. We've also tried logging events, but when we stop receiving the messages, we do not get a single event fired from sqs-consumer.

Any ideas? Thanks!

Description

1. Update the peer and development dependencies to AWS SDK v3.
2. Update all SQS client method calls to use updated syntax
3. Update error types to match new error type thrown by client
4. Update tests to run with the new API calls and error structure

Motivation and Context

1. This change updates the library to use the newest major version of the AWS SDK, which modularizes the entire SDK and substantially changes the API. As AWS pushed developers towards the new SDK, it will be important for widely-used libraries to update functionality
2. Closes #251
3. Mainly, this was selfishly motivated. I thought we needed this update for a project I'm working on. Then we pivoted and decided we didn't need it, but not before I had done nearly all the work to update the lib. I decided to finish my work and share the results in case they are useful to others. It's a small enough lib that if others needed this functionality locally, they could reasonably just copy/paste the changes into their project.

Types of changes

• [ ] Bug fix (non-breaking change which fixes an issue)
• [x] New feature (non-breaking change which adds functionality)
• [x] Breaking change (fix or feature that would cause existing functionality to change)

Checklist:

• [x] My code follows the code style of this project.
• [x] My change requires a change to the documentation.
• [x] I have updated the documentation accordingly.
• [x] I have read the CONTRIBUTING document.
• [ ] I have added tests to cover my changes.
• [x] All new and existing tests passed.

Notes on the Checklist:

1. I didn't really add tests because technically the behavior shouldn't change with this PR. However, I did substantial updates to the tests to make them work. The test coverage should not have changed at all, but I can add tests if desired.
2. I know the contributing guidelines recommend waiting for issue acknowledgement before working on a PR. However, as indicated above, I thought I needed this work done for a project and by the time I realized I didn't need it, it was already nearly done. Please note: if you don't want to use these changes, I am 100% OK with that. I am just posting them here for the community if others find themselves wanting/needing these changes, or if the library maintainers want to use this as a starting place.

Testing

In addition to updating the unit tests, I tested basic functionality locally using this script. I

// To test this against live resources, you must provision and AWS SQS queu first.
// These commands assume you have the AWS CLI installed and configured
//
// Create a queue
//    queue=$(aws sqs create-queue --queue-name my-queue --query "QueueUrl")
// Send a message to the queue
//    aws sqs send-message --queue-url $queue --message-body '{"user_id": "123123123"}'
// Run the consumer app locally
//    npx ts-node test-consumer.ts
// Delete queue when testing is complete
//    aws sqs delete-queue --queue-url $queue

import { Consumer, Message } from './index'
import { SQSClient } from '@aws-sdk/client-sqs'

const ACCOUND_ID = '123456789'
const QUEUE_NAME = 'my-queue'
const REGION = 'us-west-2'
const queueUrl = `https://sqs.${REGION}.amazonaws.com/${ACCOUND_ID}/${QUEUE_NAME}`

const sqs = new SQSClient({ region: REGION, credentialDefaultProvider: () => () => Promise.resolve({ accessKeyId: 'accessKeyId', secretAccessKey: 'secretAccessKey', sessionToken: 'sessionToken' }) })

export const handleMessage = async (message: Message) => {
  try {
    const parsed = JSON.parse(message.Body)
    console.log(parsed)
  } catch (e) {
    console.error('Parse failure!', e)
  }
}

const app = Consumer.create({
  queueUrl,
  handleMessage,
  sqs,
})

app.on('error', (err) => {
  console.error(err.message)
})

app.on('processing_error', (err) => {
  console.error(err.message)
})

app.on('timeout_error', (err) => {
  console.error(err.message)
})

app.start()

Is it possible to only delete some of the messages received in a batch? I understand that using handleMessage in combination with a batchSize greater than 1 results in them being processed one at a time but handleMessageBatch seems to be all or nothing. For example, if I had an API that could save multiple messages at a time and return a list of those that were successfully saved, could I use that to delete only the successfully saved messages from the queue?

Does the consumer continue to poll while processing incoming messages? I'm concerned whether it doesn't work sequentially and pausing the polling while handling messages as this can starve the server (if for example many messages wait on the queue and the server has capacity to handle many more messages than exist in the batch)

Description

When providing a handleMessageBatch function, code currently allows ack all messages or throw an error which will reject all messages. The fix allows the user to return an array of successful messages which will be deleted from the queue, the rest of the messages will return to the queue

Motivation and Context

Added for personal use and response to issue 245 https://github.com/bbc/sqs-consumer/issues/245

Types of changes

• [ ] Bug fix (non-breaking change which fixes an issue)
• [X] New feature (non-breaking change which adds functionality)
• [ ] Breaking change (fix or feature that would cause existing functionality to change)

Checklist:

• [X] My code follows the code style of this project.
• [X] My change requires a change to the documentation.
• [X] I have updated the documentation accordingly.
• [X] I have read the CONTRIBUTING document.
• [X] I have added tests to cover my changes.
• [X] All new and existing tests passed.

Based on a code from master branch:

public start(): void {
    if (this.stopped) {
      debug('Starting consumer');
      this.stopped = false;
      this.poll();
    }
  }

  public stop(): void {
    debug('Stopping consumer');
    this.stopped = true;
  }

private poll(): void {
    if (this.stopped) {
      this.emit('stopped');
      return;
    }

    /* tldr */
      }).then(() => {
        setTimeout(this.poll, currentPollingTimeout);
      }).catch((err) => {
        this.emit('error', err);
      });
  }

If I would do something like:

consumer.start();
consumer.stop();
consumer.start();

Wouldn't this actually invoke two times:

this.receiveMessage(receiveParams)
      .then(this.handleSqsResponse)

Which would essentially lead toward double polling and quota usage? Shouldn't there be some usage of clearTimeout() method in stop() if polling is already scheduled via setTimeout() so that we interrupt poll() from being invoked?

I was thinking to use this approach where I would stop a consumer when the queue is empty and in certain cases, if it's not running to start polling again, although I believe that this could be an edge-case and a bug.

Hi,

My app seems to consume and process messages simultaneously, even though I am using handleMessage option not handleMessageBatch().

It is really important to my application that messages are consumed one-at-a-time

Description

I just updated from 3.8.0 -> 5.0.0 and experienced ever increasing memory usage.

The update was reverted with release 034d565f with basically the following changes.

...
+ "sqs-consumer": "^3.8.0",
- "sqs-consumer": "^5.0.0",
...

+const Consumer = require('sqs-consumer')
-const { Consumer } = require('sqs-consumer')
...

+const handler = async (data, done) => {
-const handler = async (data) => {
  try {
      await doStuff(data)
  } catch (error) {
    // log error
+    return done(error)
-    throw error
  }
+  done()
}

Steps to reproduce

Don't really have a good way to reproduce.

Operating system

Running on a Standard-1X dyno in Heroku, node 8.14.1

Would appreciate some help in figuring out this issue. I'm running 7 workers in parallel using pm2. After a few minutes, only one worker continues to process messages. A few minutes after that, no workers are receiving messages. Message Received and Message Processed are printed by all workers, then just a few workers, then one, and then none.

Here are my settings and relevant code.

Queue Settings

Default Visibility Timeout: 90 seconds Message Retention Period: 4 days Maximum Message Size: 256kb Delivery Delay: 0s Receive Message Wait Time: 10s

Code

import 'babel-polyfill'

/** Libraries */
import SQSConsumer from 'sqs-consumer'
import serializeError from 'serialize-error'

/** Internal */
import IndexingService from './services/indexing.service'
import logger from './utilities/logger'
import ravenClient from './utilities/raven'

const indexingService = new IndexingService()

const startProcess = async () => {
    const app = SQSConsumer.create({
        queueUrl: 'https://sqs.us-east-1.amazonaws.com/your-queue-here',
        waitTimeSeconds: 5,
        visibilityTimeout: 90,
        handleMessage: async (message, done) => {
            try {
                const parsedBody = JSON.parse(message.Body)
                await indexingService.startProcess(parsedBody)

                done()
            }
            catch (e) {
                logger.error({
                    'Error': serializeError(e)
                })

                if(process.env.NODE_ENV == 'production') {
                    //ravenClient.captureException(e)
                }

                done(e)
            }
        }
    })

    app.on('error', (e) => {
        logger.error({
            'Error': serializeError(e)
        })

        console.log(e);
    });

    app.on('processing_error', (e, message) => {
        logger.error({
            'Error': serializeError(e)
        })

        console.log(message);
    });

    app.on('stopped', () => {
        console.log('Worker Stopped.');
    });

    app.on('message_received', () => {
        console.log('Message Received.');
    });

    app.on('message_processed', () => {
        console.log('Message Processed.');
    });

    app.start()
}

startProcess()

From the documentation of sqs-consumer, I understood that a message from the queue would be deleted only if "done()" is called in handleMessage. I have a usecase that I want the message to be left in the queue after being processed. But when the run the following code, node exits after processing a single message and only starts polling only if I call done()

Is this expected?

var Consumer = require('sqs-consumer');
var AWS = require('aws-sdk');
AWS.config.loadFromPath('../config/aws-config.json');
var app = Consumer.create({
  queueUrl: 'https://sqs.us-east-1.amazonaws.com/123/myQ',
  waitTimeSeconds: 5,
  handleMessage: function (message, done) {
    console.log(message);
    //done();
  },
});
app.on('error', function(err) {
    console.log(err.message);
});
app.start();

Describe the bug

When processing the message batch, the processMessageBatch method does not change visibility timeout process failed messages (when terminateVisibilityTimeout is true)

Your minimal, reproducible example

...

Steps to reproduce

For example, I consuming batch 10 messages with handleMessageBatch is:

for (const message of messages) {
    if (message.Body) {
       await handle(JSON.parse(message.Body);
    }
}

Simulate it:

- Message 1: Success
- Message 2: Success
- Message 3: Failed -> throw an error

Current behavior: Not change the visibility timeout message from 3 to 10

Expected behavior

• Visibility timeout of message 3 to 10 should be changed to 0

How often does this bug happen?

None

Screenshots or Videos

No response

Platform

...

Package version

v6.1.0

AWS SDK version

No response

Additional context

No response

Code Climate is currently reporting maintainability of sqs-consumer at the level of C, with some valid concerns.

https://codeclimate.com/github/BBC/sqs-consumer

Alongside what it mentions, the code is unstructured, functions are randomly placed it it's hard to see batch and normal functions alongside each other.

We should do this before we take on further new features.

Problem

There have been a number of requests that sqs-consumer should abort any existing requests when .stop() is called, as in some implementations this is expected behaviour.

Solution

In order to support existing implementations, sqs-consumer should provide new functionality that also aborts existing requests when .stop() is called.

The suggested solution would be to add an option to the stop function that will allow users to tell us to abort any requests to AWS SQS, this would be implemented like so:

consumer.stop({abort: true})

To support backwards compatibility, this option would default to false.

If the option is true, we should implement and use the AbortController to abort SQS requests as discussed here:

https://aws.amazon.com/blogs/developer/abortcontroller-in-modular-aws-sdk-for-javascript/

More documentation is also available here:

https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-sqs/interfaces/abortcontroller.html

I don't know if we will use the aws-sdk for this though, as it would require an additional dependency that people would have to add. We can potentially use the Node implementation instead:

https://nodejs.org/api/globals.html#globals_class_abortcontroller

Question I didn't mark this as a bug report because I don't know if this is a bug or if I am using this package incorrectly. I have noticed a bug where sqs consumer possibly misses the messages in the queue. Around an year ago, it was perfectly fine and this started to happen before a few months.

Here's my code Sending messages into SQS Queue:

AWS.config.update({
  region: config.aws.sqs.region,
  accessKeyId: config.aws.sqs.access_key,
  secretAccessKey: config.aws.sqs.secret_key
});

var sqs = new AWS.SQS({ apiVersion: '2012-11-05' });

module.exports.publishMessage = (txHash, eventLogs) => {
  return new Promise(async (resolve, reject) => {
    try {

      let sent_count = 0;

      logger.info(null, `Publishing events ${eventLogs.length}`)

      eventLogs.forEach((event) => {
        event.transaction_id = txHash;
        const param = {
          MessageBody: JSON.stringify(event),
          MessageGroupId: uuid.v4(),
          MessageDeduplicationId: uuid.v4(),
          QueueUrl: config.aws.sqs.node_consumer_queue_url
        }
        logger.info(null, `Sending events ${JSON.stringify(param)}`)

        sqs.sendMessage(param, function (err, data) {
          if (err) logger.error(null, err); // an error occurred
          else {
            logger.info(null, `Published: ${JSON.stringify(data)}`);

            // Debug purposes
            sent_count += 1;
            console.log(sent_count, "Sent count!");
            resolve();
          }
        });
      });
    } catch (e) {
      console.log('e', e);
      resolve();
    }
  })
}

Receiving messages from the queue

AWS.config.update({
	region: config.aws.sqs.region,
	accessKeyId: config.aws.sqs.access_key,
	secretAccessKey: config.aws.sqs.secret_key
});

let total_count = 0;

// Create our consumer
const app = Consumer.create({
	queueUrl: queueUrl,
	handleMessageBatch: async (messages) => {
		total_count += messages.length;
		return new Promise(async (resolve, reject) => {
			async.each(messages, async function (message, callback) {
				// console.log(message);
				console.log(total_count, "Received count!");

				resolve();
			});
			console.log("Execution complete!!");

		});
	},
	sqs: new AWS.SQS({
		httpOptions: {
			agent: new https.Agent({
				keepAlive: true
			})
		}
	}),
	batchSize: 10,
	waitTimeSeconds: 5
});

app.on('error', (err) => {
	logger.error(null, `Error in consumer: ${JSON.stringify(err)}`, err, err.message);
});

app.on('message_received', (msg) => {
	logger.info(null, `SQS Message Received`)
});

Sending messages in queue is working correctly whereas receiving is the part where works incorrects. When I sent 20 messages, it received 12, upon sending 10 messages, it received 8. Something weird is going on.

screenshots

Additional context Version: Latest

Question

I seem to be having an issue. Let's say that I have 5 messages in my queue and I start the poller. The first message is pulled and work begins on handling it. During processing, I have a shell command that executes that can take quite awhile to return and I need the poller to wait until the previous message is completed entirely before it moves on. Unfortunately, it appears that it is only waiting the pollingWaitTimeMs time and then it is processing the next message before the previous message is done. It is even executing the message_processed callback even though the shell command has not been completed.

screenshots

NA

Additional context

Basically, I am downloading an image from an object storage service like S3 (but not S3) and saving it locally to my server. Once saved, I am using that image to create map tiles using gdal2tiles.py however, this command can take sometime to process and finish. I need the poller to wait for this command to finish before it grabs the next message in the queue.

Code Sample

// Watcher Setup
const watcher = Consumer.create({
  queueUrl: global.config.msgsrvc.url,
  region: 'us-east-1',
  pollingWaitTimeMs: 60000,
  messageAttributeNames: ['companyId', 'farmId', 'fieldId', 'flightId', 'pipelineId', 'type', 'uuid', 'bucket', 'path', 'filename'],
  handleMessage: async(message) => {
    console.log("Processing Message...")

    await Pipeline.update({
      status: 2,
      startedAt: db.sequelize.fn('NOW')
    }, {
      where: {
        id: message.MessageAttributes.pipelineId.StringValue,
        uuid: message.MessageAttributes.uuid.StringValue
      }
    }).then(async() => {
      console.log("Downloading image from Wasabi...")

      let directory = `./tmp/${message.MessageAttributes.uuid.StringValue}-${message.MessageAttributes.type.StringValue}`

      fs.mkdir(directory, async() => {
        await storage.doDownload(
          message.MessageAttributes.bucket.StringValue,
          message.MessageAttributes.path.StringValue,
          message.MessageAttributes.filename.StringValue,
          directory
        ).then(async() => {
          console.log("Creating tiles...")

          let dirTiles = `${directory}/tiles`

          fs.mkdir(dirTiles, () => {
            let command = `gdal2tiles.py -z 1-20 --processes=10 ${directory}/${message.MessageAttributes.filename.StringValue} ${dirTiles}`

            // cmd.get(command, (data) => {
            //   console.log("Tiles created successfully...")
            //   console.log("Uploading tiles to Wasabi...")
            // })

            exec(command, (error, stdout, stderr) => {
              if (error) {
                console.log(`Error creating tiles: ${error.message}`)
              }

              console.log("Tiles created successfully...")
              console.log("Uploading tiles to Wasabi...")
            });
          })
        }).catch((error) => {
          console.log("Failed to download image: ", error)
        })
      })
    }).catch((error) => {
      console.log("Error: ", error)
    });
  },
  sqs: new aws.SQS({
    credentials: creds,
    apiVersion: '2012-11-05',
    region: 'us-east-1'
  })
});

watcher.on('message_received', (message) => {
  console.log("Message Received...")
})

watcher.on('message_processed', async(message) => {
  console.log("Message Processed...")
})

watcher.on('empty', () => {
  console.log("SQS Queue Empty...")
})

watcher.on('error', async(err, message) => {
  console.error("General Error:", err);
});

watcher.on('processing_error', async(err, message) => {
  console.error("Processing Error: ", err);
});

watcher.on('timeout_error', (err) => {
  console.error("Timeout Error: ", err);
});

console.log("Starting SQS Watcher")

watcher.start();