Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTEXT-153 Stop the KafkaMessageListenerContainer

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-153

- ensure that the fetch loop is exited immediately after the listener container is stopped;
- do not dispatch messages for processing once components are stopped;
This commit is contained in:
Marius Bogoevici
2015-03-18 11:41:06 -04:00
committed by Artem Bilan
parent 5360aa3130
commit 0e0e8cebe3
3 changed files with 21 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/ConcurrentMessageListenerDispatcher.java
View File

@@ -113,7 +113,9 @@ class ConcurrentMessageListenerDispatcher implements Lifecycle {
}
public void dispatch(KafkaMessage message) {
delegates.get(message.getMetadata().getPartition()).enqueue(message);
if (isRunning()) {
delegates.get(message.getMetadata().getPartition()).enqueue(message);
}
}
private void initializeAndStartDispatching() {

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -397,7 +397,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle {
}
return;
}
} while (!hasErrors && !partitionsWithRemainingData.isEmpty());
} while (!hasErrors && isRunning() && !partitionsWithRemainingData.isEmpty());
}
}
if (wasInterrupted) {

32
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/QueueingMessageListenerInvoker.java
View File

@@ -116,23 +116,25 @@ class QueueingMessageListenerInvoker implements Runnable, Lifecycle {
while (this.running) {
try {
KafkaMessage message = messages.take();
try {
if (messageListener != null) {
messageListener.onMessage(message);
if (isRunning()) {
try {
if (messageListener != null) {
messageListener.onMessage(message);
}
else {
acknowledgingMessageListener.onMessage(message, new DefaultAcknowledgment(offsetManager, message));
}
}
else {
acknowledgingMessageListener.onMessage(message, new DefaultAcknowledgment(offsetManager, message));
catch (Exception e) {
if (errorHandler != null) {
errorHandler.handle(e, message);
}
}
}
catch (Exception e) {
if (errorHandler != null) {
errorHandler.handle(e, message);
}
}
finally {
if (messageListener != null) {
offsetManager.updateOffset(message.getMetadata().getPartition(),
message.getMetadata().getNextOffset());
finally {
if (messageListener != null) {
offsetManager.updateOffset(message.getMetadata().getPartition(),
message.getMetadata().getNextOffset());
}
}
}
}