Skip to content

Commit

Permalink
GH-3371: Fence child containers after ConcurrentContainer stops
Browse files Browse the repository at this point in the history
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

**Auto-cherry-pick to `3.2.x` & `3.1.x`**
  • Loading branch information
LokeshAlamuri authored Jul 31, 2024
1 parent e763de6 commit 20696f2
Show file tree
Hide file tree
Showing 3 changed files with 14 additions and 0 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,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;
Expand Down Expand Up @@ -275,6 +277,10 @@ public boolean isRunning() {
return this.running;
}

protected void setFenced(boolean fenced) {
this.fenced = fenced;
}

@Deprecated(since = "3.2", forRemoval = true)
protected boolean isPaused() {
return this.paused;
Expand Down Expand Up @@ -509,6 +515,7 @@ public final void start() {
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();
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -355,6 +355,7 @@ protected void doStop(final Runnable callback, boolean normal) {
}
}
for (KafkaMessageListenerContainer<K, V> container : this.containers) {
container.setFenced(true);
if (container.isRunning()) {
if (normal) {
container.stop(() -> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -207,6 +208,7 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
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();
Expand Down Expand Up @@ -236,6 +238,10 @@ else if (e instanceof ConcurrentContainerStoppedEvent concurrentContainerStopped
});
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
Expand Down

0 comments on commit 20696f2

Please sign in to comment.