You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by eo...@apache.org on 2021/09/12 15:45:23 UTC
[pulsar] branch branch-2.8 updated: Fix cherry-pick issue on
branch-2.8 (#11982)
This is an automated email from the ASF dual-hosted git repository.
eolivelli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.8 by this push:
new ff51a02 Fix cherry-pick issue on branch-2.8 (#11982)
ff51a02 is described below
commit ff51a02c35e77abf2f720788ff251ff300166c90
Author: lipenghui <pe...@apache.org>
AuthorDate: Sun Sep 12 23:44:37 2021 +0800
Fix cherry-pick issue on branch-2.8 (#11982)
---
.../org/apache/pulsar/broker/PulsarService.java | 10 ++++++--
.../pulsar/broker/service/BrokerService.java | 12 ++++++++--
.../broker/service/PulsarChannelInitializer.java | 15 +++++++++++-
.../broker/auth/MockedPulsarServiceBaseTest.java | 7 +++++-
.../client/api/NonDurableSubscriptionTest.java | 28 ++++++++++++++++++++++
5 files changed, 66 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 8d78d40..61ffef7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -23,6 +23,7 @@ import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
import static org.apache.pulsar.transaction.coordinator.impl.MLTransactionLogImpl.TRANSACTION_LOG_PREFIX;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -237,7 +238,7 @@ public class PulsarService implements AutoCloseable {
private ProtocolHandlers protocolHandlers = null;
private final ShutdownService shutdownService;
- private final EventLoopGroup ioEventLoopGroup;
+ protected final EventLoopGroup ioEventLoopGroup;
private MetricsGenerator metricsGenerator;
@@ -652,7 +653,7 @@ public class PulsarService implements AutoCloseable {
config, localMetadataStore, getZkClient(), bkClientFactory, ioEventLoopGroup
);
- this.brokerService = new BrokerService(this, ioEventLoopGroup);
+ this.brokerService = newBrokerService(this);
// Start load management service (even if load balancing is disabled)
this.loadManager.set(LoadManager.create(this));
@@ -1675,4 +1676,9 @@ public class PulsarService implements AutoCloseable {
|| topic.endsWith(MLPendingAckStore.PENDING_ACK_STORE_SUFFIX);
}
+ @VisibleForTesting
+ protected BrokerService newBrokerService(PulsarService pulsar) throws Exception {
+ return new BrokerService(pulsar, ioEventLoopGroup);
+ }
+
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index abc673a..56dbc9c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -27,6 +27,7 @@ import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
import static org.apache.pulsar.broker.cache.LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT;
import static org.apache.pulsar.broker.web.PulsarWebResource.joinPath;
import static org.apache.pulsar.common.events.EventsTopicNames.checkTopicIsEventsNames;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
@@ -240,6 +241,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
public static final String PRODUCER_NAME_GENERATOR_PATH = "/counters/producer-name";
private final BacklogQuotaManager backlogQuotaManager;
+ private PulsarChannelInitializer.Factory pulsarChannelInitFactory = PulsarChannelInitializer.DEFAULT_FACTORY;
private final int keepAliveIntervalSeconds;
private final PulsarStats pulsarStats;
@@ -418,7 +420,8 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
ServiceConfiguration serviceConfig = pulsar.getConfiguration();
- bootstrap.childHandler(new PulsarChannelInitializer(pulsar, false));
+ bootstrap.childHandler(
+ pulsarChannelInitFactory.newPulsarChannelInitializer(pulsar, false));
Optional<Integer> port = serviceConfig.getBrokerServicePort();
if (port.isPresent()) {
@@ -435,7 +438,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
Optional<Integer> tlsPort = serviceConfig.getBrokerServicePortTls();
if (tlsPort.isPresent()) {
ServerBootstrap tlsBootstrap = bootstrap.clone();
- tlsBootstrap.childHandler(new PulsarChannelInitializer(pulsar, true));
+ tlsBootstrap.childHandler(pulsarChannelInitFactory.newPulsarChannelInitializer(pulsar, true));
try {
listenChannelTls = tlsBootstrap.bind(new InetSocketAddress(
pulsar.getBindAddress(), tlsPort.get())).sync()
@@ -2702,4 +2705,9 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
public long getPausedConnections() {
return pausedConnections.longValue();
}
+
+ @VisibleForTesting
+ public void setPulsarChannelInitializerFactory(PulsarChannelInitializer.Factory factory) {
+ this.pulsarChannelInitFactory = factory;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
index f5f2be9..6b70f5c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.service;
import static org.apache.bookkeeper.util.SafeRunnable.safeRun;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.common.annotations.VisibleForTesting;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
@@ -129,7 +130,7 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
// ServerCnx ends up reading higher number of messages and broker can not throttle the messages by disabling
// auto-read.
ch.pipeline().addLast("flowController", new FlowControlHandler());
- ServerCnx cnx = new ServerCnx(pulsar);
+ ServerCnx cnx = newServerCnx(pulsar);
ch.pipeline().addLast("handler", cnx);
connections.put(ch.remoteAddress(), cnx);
@@ -144,4 +145,16 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
}
});
}
+
+ @VisibleForTesting
+ protected ServerCnx newServerCnx(PulsarService pulsar) throws Exception {
+ return new ServerCnx(pulsar);
+ }
+
+ public interface Factory {
+ PulsarChannelInitializer newPulsarChannelInitializer(PulsarService pulsar, boolean enableTLS) throws Exception;
+ }
+
+ public static final Factory DEFAULT_FACTORY =
+ (pulsar, tls) -> new PulsarChannelInitializer(pulsar, tls);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index fb23eb3..c6ab18d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -288,7 +288,7 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
protected PulsarService startBrokerWithoutAuthorization(ServiceConfiguration conf) throws Exception {
conf.setBrokerShutdownTimeoutMs(0L);
- PulsarService pulsar = spy(new PulsarService(conf));
+ PulsarService pulsar = spy(newPulsarService(conf));
setupBrokerMocks(pulsar);
beforePulsarStartMocks(pulsar);
pulsar.start();
@@ -297,6 +297,11 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
return pulsar;
}
+
+ protected PulsarService newPulsarService(ServiceConfiguration conf) throws Exception {
+ return new PulsarService(conf);
+ }
+
protected void setupBrokerMocks(PulsarService pulsar) throws Exception {
// Override default providers with mocked ones
doReturn(mockZooKeeperClientFactory).when(pulsar).getZooKeeperClientFactory();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
index 7f1f25b..bef6274 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
@@ -65,6 +65,34 @@ public class NonDurableSubscriptionTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ @Override
+ protected PulsarService newPulsarService(ServiceConfiguration conf) throws Exception {
+ return new PulsarService(conf) {
+
+ @Override
+ protected BrokerService newBrokerService(PulsarService pulsar) throws Exception {
+ BrokerService broker = new BrokerService(this, ioEventLoopGroup);
+ broker.setPulsarChannelInitializerFactory(
+ (_pulsar, tls) -> {
+ return new PulsarChannelInitializer(_pulsar, tls) {
+ @Override
+ protected ServerCnx newServerCnx(PulsarService pulsar) throws Exception {
+ return new ServerCnx(pulsar) {
+
+ @Override
+ protected void handleFlow(CommandFlow flow) {
+ super.handleFlow(flow);
+ numFlow.incrementAndGet();
+ }
+ };
+ }
+ };
+ });
+ return broker;
+ }
+ };
+ }
+
@Test
public void testNonDurableSubscription() throws Exception {
String topicName = "persistent://my-property/my-ns/nonDurable-topic1";