[knowprasad]

Description

In io.kestra.plugin.aws.sqs.Consume, when maxDuration was configured and the target SQS queue was initially empty on the very first receive poll (count.get() == 0), the task would exit immediately because the fallback return count.get() == 0; evaluated to true. This caused the polling loop to ignore the remainder of maxDuration.

This PR ensures that when maxDuration is configured and present, ended() checks whether the duration has actually elapsed before terminating. The count.get() == 0 fallback is only used when maxDuration is not configured, maintaining existing behavior for maxRecords-only tasks.

Changes

• Updated ended() in Consume.java so that if maxDuration is present, it returns whether the duration has elapsed instead of dropping through to count.get() == 0.
• Added unit tests in ConsumeUnitTest:
  • ended_maxDuration_emptyQueue_notElapsed_returnsFalse: verifies that an initially empty queue does not terminate early when within maxDuration.
  • ended_maxDuration_emptyQueue_elapsed_returnsTrue: verifies that polling terminates once maxDuration expires.
  • ended_onlyMaxRecords_emptyQueue_returnsTrue: verifies backward compatibility when only maxRecords is configured.
  • run_withMaxDuration_waitsAcrossEmptyPollsUntilMessageArrives: verifies that run() continues polling across empty receives until a message arrives.

Closes #852.

[fdelbrayelle]

Kestra Plugin Code Review

Business Requirements: met

Matches #852: with maxDuration set, an empty first poll no longer ends the task early, and the maxRecords-only path keeps its old behavior (stops after the first empty poll).

Kestra Guidelines: 2 findings

• 🟡 Consume.java:90 [Documentation] maxDuration description does not mention the wait (not anchored to a changed line)
• 🟢 ConsumeUnitTest.java:134 [Guidelines] Explicit local types instead of var

Not an issue: ended() is now package-private so tests can call it. It is minimal and justified by the unit tests, so no finding.

Doc suggestion (🟡 Documentation, Consume.java:90): the maxDuration description is still "Stop after this duration elapses." The behavior is now explicit: when maxDuration is set, the task keeps polling an empty queue until the duration elapses (or maxRecords is reached). Worth saying so in the description, for example "Keep polling until this duration elapses, even if the queue is empty, or until maxRecords is reached."

Security (OWASP Top 10:2025 + KPS): 0 findings

✅ No issues found. Only two Java files changed; no build, CI, dependency or .github/ changes.

Performance: 0 findings

✅ No issues found. Polling cadence is unchanged (1s long poll plus 100ms sleep), bounded by maxDuration.

Verdict: APPROVE; recommendation only, posted as COMMENT, a maintainer decides.

Confidence: 94% (100 - 5 Suggestion - 1 Nit)

Note: tests were not run by this review (static review only); CI on the head was still queued/awaiting approval at review time.

[fdelbrayelle]

🟢 [Guidelines] Explicit local types instead of var
Problem: New tests declare locals such as Consume task, AtomicInteger count, ZonedDateTime start and SqsClient mockClient with explicit types where the type is obvious from the right-hand side.
Fix: Use var for these locals (the var runContext lines already do).

[knowprasad]

Thanks for the review and suggestions @fdelbrayelle! Updated the maxDuration docstring to clarify that it keeps polling an empty queue, and switched to explicit local types across the unit tests.

[fdelbrayelle]

QA report: SQS Consume maxDuration on empty first poll

Commit: 54f06ee (7-char short sha, the full sha is blocked by the secrets hook)
Edition: OSS | Docker tag: v1.3.44 (JAR also run on v2.0.7; the screenshots are from v2.0.7)
Instance: http://localhost:18860 (local, left running for manual review, docker rm -f qa-aws860-kestra to stop)
Screenshots: https://claude.ai/artifact/FaDBG5Rk1Rv8mqYTxUfqVb
Confidence: 100% (4 PASS of 4 scenarios = 100%, no deductions; non-regression flows S2 and S3 ran)

Emulator: Floci SQS (local, 512 MB), queues qa-aws860-*, dummy credentials, no real AWS. Compared with a JAR built from main on the same image.

Flow	PR head (1.3.44 / 2.0.7)	main	Result
S1 maxDuration 12s, empty queue, 1 message sent at 5s	SUCCESS, 13.5s / 13.4s, count 1	SUCCESS in 1.9s, count 0 (bug from #852 reproduced)	PASS
S2 maxRecords 5 only, empty queue	SUCCESS in 1.2s, count 0	same	PASS (non-regression)
S3 maxRecords 3 + maxDuration 20s, 3 messages loaded	SUCCESS in 0.4s, count 3	same	PASS (non-regression)
S4 maxDuration 10s, empty queue, no message	SUCCESS, 11.2s / 11.1s, count 0	SUCCESS in 1.2s	PASS

Screenshots: S1 Gantt, S1 outputs, S2 Gantt, S3 Gantt, S4 Gantt.

Flows

All four share this shape; only id, the queue suffix and the extra properties differ.

Flow YAML (S1)

id: qa_s1_maxduration_empty_late_msg
namespace: qa.aws860
tasks:
  - id: consume
    type: io.kestra.plugin.aws.sqs.Consume
    accessKeyId: test
    secretKeyId: test
    region: us-east-1
    endpointOverride: http://host.docker.internal:14566
    queueUrl: http://host.docker.internal:14566/000000000000/qa-aws860-empty
    maxDuration: PT12S

S2: queue mr, maxRecords: 5. S3: queue full, maxRecords: 3, maxDuration: PT20S. S4: queue idle, maxDuration: PT10S.

Logs and outputs: no warnings or errors in the successful runs. S1 output file contains the late message (late-msg). On 2.0.7 the task outputs show as a file object in the UI.

No new findings (no HIGH or MEDIUM). Not covered: real AWS SQS (not testable here).