Application receiving ContainerStoppedEvent even though active MessageListenerContainer instances are processing messages.

[LokeshAlamuri]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

3.1.3

Application: SpringBoot application uses spring-kafka module to consume and process the messages. MessageListenerContainer concurrency is 4. CommonContainerStoppingErrorHandler is chosen to stop the Container on any failures. Assume it would take 60 seconds to complete processing any message.

Scenario: One of the MessageListenerContainer instace throws a fatal exception. CommonContainerStoppingErrorHandler would stop the MessageListenerContainerContainer. At this point, org.springframework.kafka.event.ContainerStoppedEvent is published 5 times (concurrency + maincontainer), eventhough MessageListenerContainer instances are processing messages.

Actually, it would take some time to complete processing of the messages in other MessageListenerContainer instaces.

Problem:

1. As of now there is no exact ApplicationListener that is published after the actual spring-kafka component is closed.
2. org.springframework.kafka.event.ContainerStoppedEvent is sent for all the MessageListenerContainer instances and the main container. Because of this, from the application side we have to maintain some state to have count of how many events are received. I feel here, publishing one event is sufficient here once the main container is actually stopped. For MessageListenerContainer instances any way org.springframework.kafka.event.ConsumerStoppedEvent is published with the proper reason.

Request: Consider sending a single org.springframework.kafka.event.ContainerStoppedEvent by tracking the closed MessageListenerContainer instances count after it receives notification using 'childStopped' method.

Sample

https://github.com/LokeshAlamuri/SpringBootKafkaConsumerDemo/

[artembilan]

Looks like you have raised another issue with similar problem: #3338.

I wonder what is the difference between them?
What has led you to raise a new one since looks like both of them talk about childStopped.

Thanks

[LokeshAlamuri]

Looks like you have raised another issue with similar problem: #3338.

I wonder what is the difference between them? What has led you to raise a new one since looks like both of them talk about childStopped.

Thanks

Yes, Both problems root cause is same. But, there is some difference between them. One is with API and the other is with ApplicationListener. Incase of the current issue, I have also requested to verify if we really need to send ContainerStoppedEvent every time MessageListener container instance is stopped. For better clarity, I have raised two different issues.

[sobychacko]

HI @LokeshAlamuri, are you thinking about working on a PR for this?

[LokeshAlamuri]

Could you please suggest if the issue I have reported is a valid one. I have identified this during the code walkthrough.

[sobychacko]

So, it looks like when the fatal exception is thrown by one container, the ContainerStoppedEvent is published on all the child containers even though they are in the middle of processing. I think waiting until all the containers have finished processing is reasonable. We will run your sample app and verify it. I think we can also probably tackle the other issue you mentioned in the context of this one since, as @artembilan mentioned above, these two issues may be related. We can look into a solution, but please let us know if you want to work on it on your end.

[sobychacko]

@LokeshAlamuri We had an internal discussion on this issue, and we think it is better to introduce a new abstraction for the stopped event of concurrent containers. That way, we don't introduce any regression for the existing behavior. Let us know if you see any issues with that idea.

[LokeshAlamuri]

@LokeshAlamuri We had an internal discussion on this issue, and we think it is better to introduce a new abstraction for the stopped event of concurrent containers. That way, we don't introduce any regression for the existing behavior. Let us know if you see any issues with that idea.

It is a good idea. It would minimize the required changes greatly. But, there is one issue. But after this change, there would be no much differnce between the ContainerStoppedEvent and ConsumerStoppedEvent. Another abstraction is added for the same purpose. This could further raise questions among the Spring users. Please consider if we could deprecate ContainerStoppedEvent in favour of new proposed abstraction and we can finally remove this event during the major release.

Or else, code changes could be made to ensure ContainerStoppedEvent is only sent once after the exit of all containers. I need to further analyze on what could be the possible changes for this approach. Testing is required for this approach.

[LokeshAlamuri]

So, it looks like when the fatal exception is thrown by one container, the ContainerStoppedEvent is published on all the child containers even though they are in the middle of processing. I think waiting until all the containers have finished processing is reasonable. We will run your sample app and verify it. I think we can also probably tackle the other issue you mentioned in the context of this one since, as @artembilan mentioned above, these two issues may be related. We can look into a solution, but please let us know if you want to work on it on your end.

Thank you for considering the issue I have reported. Could you please provide the timeline before which you are expecting me to complete the code changes. Recently, I am assigned a new project. It is bit difficult for me to work on this issue at fulltime. Would you consider if I could complete it in next 2 to 3 months. It is ok for me, if you are planning assign this task to others. I will involve in the PR discussions.

[artembilan]

@LokeshAlamuri ,
It is totally fine that you cannot contribute the fix at the moment.
We as project owners will take care about the fix which should make it into the current 3.3 version.
The feedback you give is fully enough to justify as a contribution.
So, thank you for your help making this project better!

Now on the matter of the issue.

The problem with deprecating and removing the ContainerStoppedEvent is with the fact that we don't know how users rely on it.
Plus pay attention that its contract is:

	/**
	 * Construct an instance with the provided source and container.
	 * @param source the container instance that generated the event.
	 * @param container the container or the parent container if the container is a child.
	 * @since 2.2.1
	 */
	public ContainerStoppedEvent(Object source, Object container) {

So, it might be useful in some use-case to track all the containers stopped.
Also, when the top-level event is emitted for the ConcurrentMessageListenerContainer, those source and container are identical.
Therefore what you are asking for is even possible right now when you check for source == container in your event listener.
Plus there is slight difference between ContainerStoppedEvent and ConsumerStoppedEvent. The later might be emitted even if its container is still active.

With this new ConcurrentContainerStoppedEvent we just try to come up with a convenient API to react for that specific state.
Nothing more should be broken so far.

Does that make sense to you?

[sobychacko]

Ya, ConcurrentContainerStoppedEvent will be a new abstraction that won't affect any existing events published. We can work on it @LokeshAlamuri, and when the PR is ready, we will let you know so you can review it. Thanks!

[LokeshAlamuri]

Ya, ConcurrentContainerStoppedEvent will be a new abstraction that won't affect any existing events published. We can work on it @LokeshAlamuri, and when the PR is ready, we will let you know so you can review it. Thanks!

1. Adding new abstraction is a good idea. It is not having any side effects.
2. ContainerStopped Event is bit misleading. It is making the applicaiton developer to assume that contaier is already stopped. But, actually container stopping is initiated. It is ok, if you suggest to keep the changes related to this event as before.

I have done some initial changes regarding the issue I have reported. These changes could help in understanding the issue. Please review and give your comments.

https://github.com/spring-projects/spring-kafka/compare/main...LokeshAlamuri:spring-kafka:GH-3340?expand=1