[improve][client] Terminate consumer.receive() when consumer is closed #24550

3pacccccc · 2025-07-23T15:10:31Z

Motivation

Currently, when consumer.receive() is called, it blocks indefinitely even when the consumer is being closed. This can lead to threads hanging indefinitely in applications that need to gracefully shutdown consumers. The issue can be reproduced with:

Thread thread = new Thread(() -> {
    try {
        // sleep 0.1 second to close consumer to ensure consumer.receive() is triggerd
        Thread.sleep(1000);
        consumer.close();
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
});
thread.start();
// hangs forever
consumer.receive();

in fact, this could happen in many scenario as long as consumer.receive() called before consumer.close()

so, this PR aims to ensure that consumer.receive() will immediately throw an IllegalStateException when the consumer is being closed, providing clear feedback and preventing thread hangs

Modifications

1.Added termination handling in GrowableArrayBlockingQueue to:

Check termination flag in take() and poll() methods
Throw IllegalStateException with "Queue is terminated" message when terminated

2.Modified ConsumerImpl.closeConsumerTasks() to terminate the incoming message queue

3.Added tests covering:

Regular consumer close during receive()
Zero queue size consumer receives a batchMessage and another thread close consuer

Verifying this change

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: 3pacccccc#15

lhotari

Good catch! left some review comments

...mmon/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

3pacccccc · 2025-07-24T06:38:03Z

Good catch! left some review comments

thank you so such for pointing these out, I've pushed a new commit, please take a look again.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

lhotari

LGTM, good work @3pacccccc

...mmon/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java

lhotari

There's a need to handle the case where more than one thread is waiting.

3pacccccc · 2025-07-25T08:44:15Z

There's a need to handle the case where more than one thread is waiting.

IMO, because of the headLock ensures that there is at most one thread can be waiting on the condition. so, I think the scenario handle the case where more than one thread is waiting. might not be existed.
@lhotari

lhotari · 2025-07-25T11:30:22Z

IMO, because of the headLock ensures that there is at most one thread can be waiting on the condition. so, I think the scenario handle the case where more than one thread is waiting. might not be existed.
@lhotari

isNotEmpty has been created with headLock.newCondition(). When .await() is called, the lock is released. When .signal is called, the thread that was awaiting will acquire the lock again. The difference between .signal and .signalAll is that the first one will only wake up one thread. .signalAll will wake up one thread at a time.
There is a need to handle the multi-thread case.

3pacccccc · 2025-07-25T16:21:25Z

IMO, because of the headLock ensures that there is at most one thread can be waiting on the condition. so, I think the scenario handle the case where more than one thread is waiting. might not be existed.
@lhotari

isNotEmpty has been created with headLock.newCondition(). When .await() is called, the lock is released. When .signal is called, the thread that was awaiting will acquire the lock again. The difference between .signal and .signalAll is that the first one will only wake up one thread. .signalAll will wake up one thread at a time. There is a need to handle the multi-thread case.

@lhotari yes! you're right. Thank you so much for pointing this out, I've pushed a new commit, PTAL

lhotari

LGTM, good work @3pacccccc

codecov-commenter · 2025-07-25T18:30:50Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.26%. Comparing base (bbc6224) to head (b4a02da).
⚠️ Report is 1311 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #24550      +/-   ##
============================================
+ Coverage     73.57%   74.26%   +0.69%     
+ Complexity    32624    32595      -29     
============================================
  Files          1877     1876       -1     
  Lines        139502   146326    +6824     
  Branches      15299    16781    +1482     
============================================
+ Hits         102638   108673    +6035     
- Misses        28908    29021     +113     
- Partials       7956     8632     +676

Flag	Coverage Δ
inttests	`26.70% <50.00%> (+2.11%)`	⬆️
systests	`23.28% <50.00%> (-1.04%)`	⬇️
unittests	`73.77% <100.00%> (+0.93%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	`79.24% <100.00%> (+1.67%)`	⬆️
...n/util/collections/GrowableArrayBlockingQueue.java	`97.58% <100.00%> (+2.15%)`	⬆️

... and 1111 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

#24550) (cherry picked from commit 6ad57da)

apache#24550) (cherry picked from commit 6ad57da)

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit 4dd5b1e)

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit ba9147a)

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit 4dd5b1e)

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit ba9147a)

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit 4dd5b1e)

apache#24550)

3pacccccc added 4 commits July 23, 2025 17:37

[improve][client]cut consumer.receive() when consumer get closed

2806783

[fix][client]add test for zeroQueueConsumer

869779f

[fix][client]add test time out

48387f7

[fix][client]add checks for GrowableArrayBlockingQueue.poll

cc0d6b3

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jul 23, 2025

lhotari reviewed Jul 23, 2025

View reviewed changes

[improve][client]update

d5550da

3pacccccc added 3 commits July 24, 2025 15:14

[improve][client]return null when poll message

599456a

fix checkstyle

0da8062

update assertion

2405145

3pacccccc requested a review from lhotari July 24, 2025 11:28

update assertion

64df7c8

lhotari reviewed Jul 24, 2025

View reviewed changes

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java Outdated Show resolved Hide resolved

[fix][client]remove poolMessages checks

53e6b23

lhotari approved these changes Jul 24, 2025

View reviewed changes

lhotari requested review from BewareMyPower, codelipenghui and poorbarcode July 24, 2025 18:27

lhotari reviewed Jul 25, 2025

View reviewed changes

...mmon/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java Outdated Show resolved Hide resolved

lhotari requested changes Jul 25, 2025

View reviewed changes

[fix][client]change signal to signalAll and add multi-thread test case

b4a02da

3pacccccc requested a review from lhotari July 25, 2025 16:22

lhotari approved these changes Jul 25, 2025

View reviewed changes

lhotari merged commit 6ad57da into apache:master Jul 25, 2025
96 of 98 checks passed

lhotari added this to the 4.1.0 milestone Jul 25, 2025

lhotari added release/3.0.13 release/4.0.6 release/3.3.8 labels Jul 25, 2025

lhotari pushed a commit that referenced this pull request Jul 25, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

ba9147a

#24550) (cherry picked from commit 6ad57da)

lhotari added the cherry-picked/branch-4.0 label Jul 25, 2025

lhotari pushed a commit that referenced this pull request Jul 25, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

4dd5b1e

#24550) (cherry picked from commit 6ad57da)

lhotari added the cherry-picked/branch-3.0 label Jul 25, 2025

lhotari pushed a commit that referenced this pull request Jul 25, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

485f775

#24550) (cherry picked from commit 6ad57da)

lhotari added the cherry-picked/branch-3.3 label Jul 25, 2025

nodece pushed a commit to ascentstream/pulsar that referenced this pull request Jul 28, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

3c140b7

apache#24550) (cherry picked from commit 6ad57da)

nodece pushed a commit to ascentstream/pulsar that referenced this pull request Jul 28, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

9b2c66d

apache#24550) (cherry picked from commit 6ad57da)

priyanshu-ctds pushed a commit to datastax/pulsar that referenced this pull request Jul 28, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

7d2f50e

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit 4dd5b1e)

priyanshu-ctds pushed a commit to datastax/pulsar that referenced this pull request Jul 28, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

8dcb1b1

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit ba9147a)

This was referenced Jul 29, 2025

[fix][test]fix flaky test GrowableArrayBlockingQueueTest.testPollBlockingThreadsTermination 3pacccccc/pulsar#19

Closed

[fix][test] fix flaky GrowableArrayBlockingQueueTest.testPollBlockingThreadsTermination #24576

Merged

priyanshu-ctds pushed a commit to datastax/pulsar that referenced this pull request Jul 29, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

a1c429d

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit 4dd5b1e)

3pacccccc deleted the cutReceiveCall branch July 29, 2025 15:07

srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Jul 29, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

7f12f69

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit ba9147a)

srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Jul 30, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

39cd5e7

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit ba9147a)

srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Jul 31, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

2ef2e4b

apache#24550) (cherry picked from commit 6ad57da) (cherry picked from commit 4dd5b1e)

lhotari mentioned this pull request Aug 14, 2025

Flaky-test: ZeroQueueSizeTest.testZeroQueueGetExceptionWhenReceiveBatchMessage #24629

Closed

2 tasks

KannarFr pushed a commit to CleverCloud/pulsar that referenced this pull request Sep 22, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

eb5008c

apache#24550)

walkinggo pushed a commit to walkinggo/pulsar that referenced this pull request Oct 8, 2025

[improve][client] Terminate consumer.receive() when consumer is closed (

0f5cf04

apache#24550)

[improve][client] Terminate consumer.receive() when consumer is closed #24550

[improve][client] Terminate consumer.receive() when consumer is closed #24550

Uh oh!

Conversation

3pacccccc commented Jul 23, 2025

Motivation

Modifications

Verifying this change

Documentation

Matching PR in forked repository

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

3pacccccc commented Jul 24, 2025

Uh oh!

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

3pacccccc commented Jul 25, 2025

Uh oh!

lhotari commented Jul 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

3pacccccc commented Jul 25, 2025

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Jul 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

lhotari commented Jul 25, 2025 •

edited

Loading

codecov-commenter commented Jul 25, 2025 •

edited

Loading