[ISSUE #3707]fix thread leak with rocketmq consumer start and shutdown frequently

[willimpo]

Fixes #3707.

Motivation

when using rocketmq as eventmesh storage，and consumer groups are frequently started and shutdown.

The number of threads in the process continues to increase，and eventually led to Full GC

Modifications

In the rocketmq eventmesh storage, we overwrite the ConsumeMessageConcurrentlyService.java of rocketmq
but in the shutdown(long) method, we didn't do anything, this led to some consumer threads are not recycled when consumer was shutdown

the shutdown(long) method in ConsumeMessageConcurrentlyService.java

    public void shutdown(long awaitTerminateMillis) {

    }

where the ConsumeMessageConcurrentlyService.shutdown(long) method was invoked

DefaultMQPushConsumerImpl.java:

    public synchronized void shutdown(long awaitTerminateMillis) {
        switch (this.serviceState) {
            case CREATE_JUST:
                break;
            case RUNNING:
                // here invokes the shutdown method
                this.consumeMessageService.shutdown(awaitTerminateMillis);
                this.persistConsumerOffset();
                this.mQClientFactory.unregisterConsumer(this.defaultMQPushConsumer.getConsumerGroup());
                this.mQClientFactory.shutdown();
                log.info("the consumer [{}] shutdown OK", this.defaultMQPushConsumer.getConsumerGroup());
                this.rebalanceImpl.destroy();
                this.serviceState = ServiceState.SHUTDOWN_ALREADY;
                break;
            case SHUTDOWN_ALREADY:
                break;
            default:
                break;
        }
    }

Therefore, my modification is to supplement the logic of shutdown (long) .

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[github-actions]

Welcome to the Apache EventMesh community!!
This is your first PR in our project. We're very excited to have you onboard contributing. Your contributions are greatly appreciated!

Please make sure that the changes are covered by tests.
We will be here shortly.
Let us know if you need any help!

Want to get closer to the community?

WeChat Assistant	WeChat Public Account	Slack
		Join Slack Chat

Mailing Lists:

Name	Description	Subscribe	Unsubscribe	Archive
Users	User support and questions mailing list	Subscribe	Unsubscribe	Mail Archives
Development	Development related discussions	Subscribe	Unsubscribe	Mail Archives
Commits	All commits to repositories	Subscribe	Unsubscribe	Mail Archives
Issues	Issues or PRs comments and reviews	Subscribe	Unsubscribe	Mail Archives

[codecov]

Codecov Report

Merging #3708 (0e1fc5a) into master (36036da) will decrease coverage by 0.01%.
The diff coverage is 0.00%.

❗ Current head 0e1fc5a differs from pull request most recent head 508579b. Consider uploading reports for the commit 508579b to get more accurate results

@@             Coverage Diff              @@
##             master    #3708      +/-   ##
============================================
- Coverage     13.76%   13.76%   -0.01%     
  Complexity     1291     1291              
============================================
  Files           571      571              
  Lines         29239    29242       +3     
  Branches       2884     2884              
============================================
  Hits           4024     4024              
- Misses        24841    24844       +3     
  Partials        374      374              

Impacted Files	Coverage Δ
...pl/consumer/ConsumeMessageConcurrentlyService.java	0.00% <0.00%> (ø)

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[willimpo]

@xwm1992 pls make a review,thx

[xwm1992]

LGTM