Siddhi的SQS源轮询间隔配置无效问题求助

Hey there! Let's figure out why your SQS source settings aren't taking effect in your Siddhi app. You've configured polling.interval to 10 seconds and max.number.of.messages to 1, but you're still seeing a continuous stream of messages—here are the key fixes to check:

1. Fix the Syntax Error in Your @source Annotation #

Looking at your code, there's a syntax mistake in how you've structured the @source and @map annotations. The @source parameter list isn't properly closed before starting @map, which means Siddhi might be ignoring some of your configuration values entirely.

Incorrect code snippet:

@source(type = 'sqs', queue = '', access.key = '', secret.key = '', region = '', polling.interval = '10000', wait.time = '20', max.number.of.messages = '1', delete.messages='false', number.of.parallel.consumers = '1',
@map(type = 'json', fail.on.missing.attribute='false', enclosing.element='$.entries',
@attributes(val1 = 'val1', val2='val2')))

Corrected version:

@source(type = 'sqs', queue = '', access.key = '', secret.key = '', region = '', polling.interval = '10000', wait.time = '20', max.number.of.messages = '1', delete.messages='false', number.of.parallel.consumers = '1')
@map(type = 'json', fail.on.missing.attribute='false', enclosing.element='$.entries',
@attributes(val1 = 'val1', val2='val2'))

Notice we closed the @source parentheses before adding the @map annotation—this ensures all your source configs are parsed correctly.

2. Address Message Reconsumption Due to delete.messages='false' #

You've set delete.messages='false', which means Siddhi won't delete messages from the SQS queue after consuming them. That means every time the poll runs (every 10 seconds), it will pull the same un-deleted message again, creating the "continuous stream" effect you're seeing.

If you want messages to be processed only once, change this setting to delete.messages='true'. If you need to keep messages in the queue for other consumers, you'll need to handle deletion manually after processing.

3. Verify Parameter Units and Compatibility #

• polling.interval is set to 10000 (milliseconds), which correctly translates to 10 seconds—this part is fine.
• wait.time='20' sets the SQS long-poll wait time to 20 seconds, which is the maximum allowed value. This works with your polling interval, but keep in mind that if a message arrives during the long poll, Siddhi will receive it immediately instead of waiting the full 10 seconds.

Full Corrected App Code #

Here's your full app with the syntax fix applied (and assuming you want to delete messages after processing):

@App:name('SQS TEST')
@App:description('Description of the plan')

@source(type = 'sqs', queue = '', access.key = '', secret.key = '', region = '', polling.interval = '10000', wait.time = '20', max.number.of.messages = '1', delete.messages='true', number.of.parallel.consumers = '1')
@map(type = 'json', fail.on.missing.attribute='false', enclosing.element='$.entries',
@attributes(val1 = 'val1', val2='val2'))
define stream inStream (val1 string, val2 string);

@sink(type = 'log', prefix = 'Cafe Feed:')
define stream log_received (val1 string, val2 string);

from inStream select val1,val2 insert into log_received;

Try these changes out—your polling interval and message limit should start working as expected once the syntax is fixed and message deletion is handled appropriately.

内容的提问来源于stack exchange，提问作者muqy