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

Container pause/resume improvements (#299)

Browse Source 
- When container is paused, ensure that we are not spinning
  CPU cycles by constantly trying to poll in a loop. When
  pausing the container, the polling waits until a condition
  is signaled by resuming.

Resolves #297
This commit is contained in:
Soby Chacko
2023-01-31 14:33:43 -05:00
committed by GitHub
parent 224eb0b3e2
commit ce96e7f5a9
2 changed files with 110 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -30,6 +30,9 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
@@ -91,6 +94,10 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
private final AtomicBoolean receiveInProgress = new AtomicBoolean();
private final Lock lockOnPause = new ReentrantLock();
private final Condition pausedCondition = this.lockOnPause.newCondition();
public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
PulsarContainerProperties pulsarContainerProperties) {
this(pulsarConsumerFactory, pulsarContainerProperties, null);
@@ -207,6 +214,14 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
consumer.resume();
}
setPaused(false);
this.lockOnPause.lock();
try {
// signal the lock's condition to continue.
this.pausedCondition.signal();
}
finally {
this.lockOnPause.unlock();
}
}
private final class Listener implements SchedulingAwareRunnable {
@@ -352,6 +367,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
Messages<T> messages = null;
List<Message<T>> messageList = null;
while (isRunning()) {
checkIfPausedAndHandleAccordingly();
// Always receive messages in batch mode.
try {
if (!inRetryMode.get() && !messagesPendingInBatch.get()) {
@@ -442,6 +458,23 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
}
private void checkIfPausedAndHandleAccordingly() {
if (isPaused()) {
// try acquiring the lock.
DefaultPulsarMessageListenerContainer.this.lockOnPause.lock();
try {
// Waiting on lock's condition.
DefaultPulsarMessageListenerContainer.this.pausedCondition.await();
}
catch (InterruptedException e) {
throw new IllegalStateException("Exception occurred trying to wake up the paused listener thread.");
}
finally {
DefaultPulsarMessageListenerContainer.this.lockOnPause.unlock();
}
}
}
private Observation newObservation(Message<T> message) {
if (this.observationRegistry == null) {
return Observation.NOOP;

77
spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.pulsar.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -30,8 +31,12 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.DeadLetterPolicy;
@@ -42,6 +47,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Test;
import org.springframework.pulsar.core.ConsumerTestUtils;
@@ -49,6 +55,7 @@ import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import org.springframework.test.util.ReflectionTestUtils;
/**
* @author Soby Chacko
@@ -76,6 +83,7 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container.start();
Map<String, Object> prodConfig = new HashMap<>();
prodConfig.put("topicName", "dpmlct-012");
DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
@@ -87,6 +95,75 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
pulsarClient.close();
}
@Test
void containerPauseAndResumeFeatureUsingWaitAndNotify() throws Exception {
Set<String> topics = Collections.singleton("containerPauseResumeWaitNotify-topic");
Map<String, Object> config = Map.of("topicNames", topics, "subscriptionName",
"containerPauseResumeWaitNotify-sub");
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
.build();
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
config);
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
});
pulsarContainerProperties.setSchema(Schema.STRING);
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
Lock reentrantLock = new ReentrantLock();
Condition lockCondition = reentrantLock.newCondition();
Lock spyLock = spy(reentrantLock);
Condition spyCondition = spy(lockCondition);
ReflectionTestUtils.setField(container, "lockOnPause", spyLock);
ReflectionTestUtils.setField(container, "pausedCondition", spyCondition);
CountDownLatch latchOnLockInvocation = new CountDownLatch(2);
CountDownLatch latchOnUnlockInvocation = new CountDownLatch(2);
CountDownLatch latchOnAwaitInvocation = new CountDownLatch(1);
CountDownLatch latchOnSignalInvocation = new CountDownLatch(1);
doAnswer(invocation -> {
latchOnLockInvocation.countDown();
return invocation.callRealMethod();
}).when(spyLock).lock();
doAnswer(invocation -> {
latchOnUnlockInvocation.countDown();
return invocation.callRealMethod();
}).when(spyLock).unlock();
doAnswer(invocation -> {
latchOnAwaitInvocation.countDown();
return invocation.callRealMethod();
}).when(spyCondition).await();
doAnswer(invocation -> {
latchOnSignalInvocation.countDown();
return invocation.callRealMethod();
}).when(spyCondition).signal();
container.start();
container.pause();
Awaitility.await().until(container::isPaused);
container.resume();
Awaitility.await().until(() -> !container.isPaused());
assertThat(latchOnLockInvocation.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchOnUnlockInvocation.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchOnAwaitInvocation.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(latchOnSignalInvocation.await(10, TimeUnit.SECONDS)).isTrue();
container.stop();
pulsarClient.close();
}
@Test
void subscriptionInitialPositionEarliest() throws Exception {
Map<String, Object> config = new HashMap<>();