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

GH-3371: Fence child containers after ConcurrentContainer stops

Browse Source 
Fixes: #3371

Containers are not restricted from starting after ConcurrentContainer stopped or restarted. These changes would fix this issue.

* Enhancements to fence container after ConcurrentContainer stops

(cherry picked from commit 20696f2390)

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
This commit is contained in:
LokeshAlamuri
2024-07-31 22:28:38 +05:30
committed by Artem Bilan
parent c900940b34
commit 3fdf4682f2
3 changed files with 14 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
View File

@@ -123,6 +123,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
private volatile boolean running = false;
private volatile boolean fenced = false;
private volatile boolean paused;
private volatile boolean stoppedNormally = true;
@@ -274,6 +276,10 @@ public abstract class AbstractMessageListenerContainer<K, V>
return this.running;
}
protected void setFenced(boolean fenced) {
this.fenced = fenced;
}
protected boolean isPaused() {
return this.paused;
}
@@ -507,6 +513,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
if (!isRunning()) {
Assert.state(this.containerProperties.getMessageListener() instanceof GenericMessageListener,
() -> "A " + GenericMessageListener.class.getName() + " implementation must be provided");
Assert.state(!this.fenced, "Container Fenced. It is not allowed to start.");
doStart();
}
}

1
spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
View File

@@ -351,6 +351,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
}
}
for (KafkaMessageListenerContainer<K, V> container : this.containers) {
container.setFenced(true);
if (container.isRunning()) {
if (normal) {
container.stop(() -> {

6
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.kafka.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.given;
@@ -198,6 +199,7 @@ public class ConcurrentMessageListenerContainerTests {
assertThat(container.metrics()).isNotNull();
Set<KafkaMessageListenerContainer<Integer, String>> children = new HashSet<>(containers);
assertThat(container.isInExpectedState()).isTrue();
MessageListenerContainer childContainer = container.getContainers().get(0);
container.getContainers().get(0).stopAbnormally(() -> { });
assertThat(container.isInExpectedState()).isFalse();
container.getContainers().get(0).start();
@@ -220,6 +222,10 @@ public class ConcurrentMessageListenerContainerTests {
});
assertThat(overrides.get().getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)).isNull();
this.logger.info("Stop auto");
assertThat(childContainer.isRunning()).isFalse();
assertThat(container.isRunning()).isFalse();
// Fenced container. Throws exception
assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> childContainer.start());
}
@Test