Optimize conusmer pause
@@ -1028,6 +1028,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
partitionIndex -> {
String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
+ configurationData.setStartPaused(paused);
ConsumerImpl<T> newConsumer = ConsumerImpl.newConsumerImpl(client, partitionName,
configurationData, client.externalExecutorProvider(),
partitionIndex, true, subFuture,
@@ -1036,6 +1037,8 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
synchronized (pauseMutex) {
if (paused) {
newConsumer.pause();
+ } else {
+ newConsumer.resume();
}
consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
}
@@ -1055,6 +1058,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
subscribeResult.completeExceptionally(new PulsarClientException(errorMessage));
return existingValue;
} else {
+ internalConfig.setStartPaused(paused);
ConsumerImpl<T> newConsumer = ConsumerImpl.newConsumerImpl(client, topicName, internalConfig,
client.externalExecutorProvider(), -1,
true, subFuture, startMessageId, schema, interceptors,
@@ -1063,6 +1067,8 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
synchronized (pauseMutex) {
if (paused) {
newConsumer.pause();
+ } else {
+ newConsumer.resume();
}
}
return newConsumer;
@@ -1355,6 +1361,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
int partitionIndex = TopicName.getPartitionIndex(partitionName);
CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
ConsumerConfigurationData<T> configurationData = getInternalConsumerConfig();
+ configurationData.setStartPaused(paused);
ConsumerImpl<T> newConsumer = ConsumerImpl.newConsumerImpl(
client, partitionName, configurationData,
client.externalExecutorProvider(),
@@ -1363,6 +1370,8 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
synchronized (pauseMutex) {
if (paused) {
newConsumer.pause();
+ } else {
+ newConsumer.resume();
}
consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
}
