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

Polish "Consumer restart causes duplicate messages"

Browse Source 
- Make `listenerConsumerThread` reference final
This commit is contained in:
Chris Bono
2022-10-19 20:50:04 -05:00
parent 2360a9d2d1
commit c3f379e2e3
1 changed files with 3 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -83,7 +83,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
private final AbstractPulsarMessageListenerContainer<?> thisOrParentContainer;
private AtomicReference<Thread> listenerConsumerThread;
private final AtomicReference<Thread> listenerConsumerThread = new AtomicReference<>();
private final AtomicBoolean receiveInProgress = new AtomicBoolean();
@@ -138,7 +138,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
setRunning(false);
this.logger.info("Pausing this consumer.");
this.listenerConsumer.consumer.pause();
if (this.listenerConsumerThread != null) {
if (this.listenerConsumerThread.get() != null) {
// if there is a receive operation already in progress, we want to interrupt
// the listener thread.
if (this.receiveInProgress.get()) {
@@ -309,8 +309,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
@Override
public void run() {
DefaultPulsarMessageListenerContainer.this.listenerConsumerThread = new AtomicReference<>(
Thread.currentThread());
DefaultPulsarMessageListenerContainer.this.listenerConsumerThread.set(Thread.currentThread());
publishConsumerStartingEvent();
publishConsumerStartedEvent();
AtomicBoolean inRetryMode = new AtomicBoolean(false);