From 124e604a93382a5672288cb3fd5b02b8f73fee28 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Mon, 29 Sep 2025 14:28:44 +0800 Subject: [PATCH 1/8] [improve][broker]Use ServerCnxThrottleTracker, instead of modifying channel.readable directly --- .../ResourceGroupPublishLimiter.java | 9 +- .../pulsar/broker/service/AbstractTopic.java | 7 +- .../pulsar/broker/service/BrokerService.java | 8 +- .../pulsar/broker/service/Producer.java | 24 - .../service/PublishRateLimiterImpl.java | 13 +- .../pulsar/broker/service/ServerCnx.java | 35 +- .../service/ServerCnxThrottleTracker.java | 478 +++++++++++++++--- .../pulsar/broker/service/TransportCnx.java | 17 +- .../PublishRateLimiterDisableTest.java | 22 +- .../service/PublishRateLimiterTest.java | 58 ++- .../service/TopicPublishRateThrottleTest.java | 31 ++ 11 files changed, 523 insertions(+), 179 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java index fc4514db81fe6..edc723c35e94c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java @@ -21,6 +21,7 @@ import org.apache.pulsar.broker.qos.MonotonicClock; import org.apache.pulsar.broker.resourcegroup.ResourceGroup.BytesAndMessagesCount; import org.apache.pulsar.broker.service.PublishRateLimiterImpl; +import org.apache.pulsar.broker.service.ServerCnxThrottleTracker; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.PublishRate; import org.apache.pulsar.common.policies.data.ResourceGroup; @@ -30,7 +31,13 @@ public class ResourceGroupPublishLimiter extends PublishRateLimiterImpl { private volatile long publishMaxByteRate; public ResourceGroupPublishLimiter(ResourceGroup resourceGroup, MonotonicClock monotonicClock) { - super(monotonicClock); + super(monotonicClock, producer -> { + producer.getCnx().getThrottleTracker().markThrottled( + ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + }, producer -> { + producer.getCnx().getThrottleTracker().unmarkThrottled( + ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + }); update(resourceGroup); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index 3ec6f5a0cd5e6..f1b0d5eadfe82 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -21,6 +21,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; import static org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl.ENTRY_LATENCY_BUCKETS_USEC; +import static org.apache.pulsar.broker.service.ServerCnxThrottleTracker.ThrottleType; import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.MoreObjects; @@ -193,7 +194,11 @@ public AbstractTopic(String topic, BrokerService brokerService) { updateTopicPolicyByBrokerConfig(); this.lastActive = System.nanoTime(); - topicPublishRateLimiter = new PublishRateLimiterImpl(brokerService.getPulsar().getMonotonicClock()); + topicPublishRateLimiter = new PublishRateLimiterImpl(brokerService.getPulsar().getMonotonicClock(), producer -> { + producer.getCnx().getThrottleTracker().markThrottled(ThrottleType.TopicPublishRate); + }, producer -> { + producer.getCnx().getThrottleTracker().unmarkThrottled(ThrottleType.TopicPublishRate); + }); updateActiveRateLimiters(); additionalSystemCursorNames = brokerService.pulsar().getConfiguration().getAdditionalSystemCursorNames(); 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 79dffdf7aadef..378fda44c2e1c 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 @@ -339,7 +339,13 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws this.pulsar = pulsar; this.clock = pulsar.getClock(); this.dynamicConfigurationMap = prepareDynamicConfigurationMap(); - this.brokerPublishRateLimiter = new PublishRateLimiterImpl(pulsar.getMonotonicClock()); + this.brokerPublishRateLimiter = new PublishRateLimiterImpl(pulsar.getMonotonicClock(), producer -> { + producer.getCnx().getThrottleTracker().markThrottled( + ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + }, producer -> { + producer.getCnx().getThrottleTracker().unmarkThrottled( + ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + }); this.dispatchRateLimiterFactory = createDispatchRateLimiterFactory(pulsar.getConfig()); this.managedLedgerStorage = pulsar.getManagedLedgerStorage(); this.keepAliveIntervalSeconds = pulsar.getConfiguration().getKeepAliveIntervalSeconds(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java index 0784f74591ec5..9d0c10802546f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java @@ -899,30 +899,6 @@ public boolean isDisconnecting() { private static final Logger log = LoggerFactory.getLogger(Producer.class); - /** - * This method increments a counter that is used to control the throttling of a connection. - * The connection's read operations are paused when the counter's value is greater than 0, indicating that - * throttling is in effect. - * It's important to note that after calling this method, it is the caller's responsibility to ensure that the - * counter is decremented by calling the {@link #decrementThrottleCount()} method when throttling is no longer - * needed on the connection. - */ - public void incrementThrottleCount() { - cnx.incrementThrottleCount(); - } - - /** - * This method decrements a counter that is used to control the throttling of a connection. - * The connection's read operations are resumed when the counter's value is 0, indicating that - * throttling is no longer in effect. - * It's important to note that before calling this method, the caller should have previously - * incremented the counter by calling the {@link #incrementThrottleCount()} method when throttling - * was needed on the connection. - */ - public void decrementThrottleCount() { - cnx.decrementThrottleCount(); - } - public Attributes getOpenTelemetryAttributes() { if (attributes != null) { return attributes; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java index 0015f2675a2f0..096418191dc44 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.qos.AsyncTokenBucket; import org.apache.pulsar.broker.qos.MonotonicClock; @@ -42,9 +43,14 @@ public class PublishRateLimiterImpl implements PublishRateLimiter { private final AtomicInteger throttledProducersCount = new AtomicInteger(0); private final AtomicBoolean processingQueuedProducers = new AtomicBoolean(false); + private final Consumer throttleAction; + private final Consumer unthrottleAction; - public PublishRateLimiterImpl(MonotonicClock monotonicClock) { + public PublishRateLimiterImpl(MonotonicClock monotonicClock, Consumer throttleAction, + Consumer unthrottleAction) { this.monotonicClock = monotonicClock; + this.throttleAction = throttleAction; + this.unthrottleAction = unthrottleAction; } /** @@ -68,7 +74,7 @@ public void handlePublishThrottling(Producer producer, int numOfMessages, } if (shouldThrottle) { // throttle the producer by incrementing the throttle count - producer.incrementThrottleCount(); + throttleAction.accept(producer); // schedule decrementing the throttle count to possibly unthrottle the producer after the // throttling period scheduleDecrementThrottleCount(producer); @@ -136,7 +142,8 @@ private void unthrottleQueuedProducers(ScheduledExecutorService executor) { while ((throttlingDuration = calculateThrottlingDurationNanos()) == 0L && (producer = unthrottlingQueue.poll()) != null) { try { - producer.decrementThrottleCount(); + final Producer producerFinal = producer; + producer.getCnx().execute(() -> unthrottleAction.accept(producerFinal)); } catch (Exception e) { log.error("Failed to unthrottle producer {}", producer, e); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index fbfb8108846b0..25642a0b670b1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -27,6 +27,7 @@ import static org.apache.pulsar.broker.lookup.TopicLookupBase.lookupTopicAsync; import static org.apache.pulsar.broker.service.persistent.PersistentTopic.getMigratedClusterUrl; import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.ignoreUnrecoverableBKException; +import static org.apache.pulsar.broker.service.ServerCnxThrottleTracker.ThrottleType; import static org.apache.pulsar.common.api.proto.ProtocolVersion.v5; import static org.apache.pulsar.common.naming.Constants.WEBSOCKET_DUMMY_ORIGINAL_PRINCIPLE; import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH; @@ -283,7 +284,8 @@ public void incrementPublishBytes(long bytes, long maxPendingBytesPerThread) { if (maxPendingBytesPerThread > 0 && pendingBytes > maxPendingBytesPerThread && !limitExceeded) { limitExceeded = true; - cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.setPublishBufferLimiting(true)); + cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.markThrottled( + ThrottleType.IOThreadMaxBytesOfInFlightPublishing)); } } @@ -293,7 +295,8 @@ public void decrementPublishBytes(long bytes, long resumeThresholdPendingBytesPe // we resume all connections sharing the same thread if (limitExceeded && pendingBytes <= resumeThresholdPendingBytesPerThread) { limitExceeded = false; - cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.setPublishBufferLimiting(false)); + cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.unmarkThrottled( + ThrottleType.IOThreadMaxBytesOfInFlightPublishing)); } } } @@ -311,6 +314,7 @@ enum State { Start, Connected, Failed, Connecting } + @Getter private final ServerCnxThrottleTracker throttleTracker; public ServerCnx(PulsarService pulsar) { @@ -481,12 +485,12 @@ private void checkPauseReceivingRequestsAfterResumeRateLimit(BaseCommand cmd) { log.warn("[{}] Reached rate limitation", this); // Stop receiving requests. pausedDueToRateLimitation = true; - ctx.channel().config().setAutoRead(false); + getThrottleTracker().markThrottled(ThrottleType.ConnectionPauseReceivingCooldownRateLimit); // Resume after 1 second. ctx.channel().eventLoop().schedule(() -> { if (pausedDueToRateLimitation) { log.info("[{}] Resuming connection after rate limitation", this); - ctx.channel().config().setAutoRead(true); + getThrottleTracker().unmarkThrottled(ThrottleType.ConnectionPauseReceivingCooldownRateLimit); pausedDueToRateLimitation = false; } }, requestRateLimiter.getPeriodAtMs(), TimeUnit.MILLISECONDS); @@ -497,7 +501,7 @@ private void checkPauseReceivingRequestsAfterResumeRateLimit(BaseCommand cmd) { public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { if (pauseReceivingRequestsIfUnwritable && ctx.channel().isWritable()) { log.info("[{}] is writable, turn on channel auto-read", this); - ctx.channel().config().setAutoRead(true); + getThrottleTracker().markThrottled(ThrottleType.ConnectionOutboundBufferFull); requestRateLimiter.timingOpen(pauseReceivingCooldownMilliSeconds, TimeUnit.MILLISECONDS); } else if (pauseReceivingRequestsIfUnwritable && !ctx.channel().isWritable()) { final ChannelOutboundBuffer outboundBuffer = ctx.channel().unsafe().outboundBuffer(); @@ -511,6 +515,7 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exceptio PAUSE_RECEIVING_LOG.debug("[{}] is not writable, turn off channel auto-read", this); } } + getThrottleTracker().unmarkThrottled(ThrottleType.ConnectionOutboundBufferFull); ctx.channel().config().setAutoRead(false); } ctx.fireChannelWritabilityChanged(); @@ -3399,7 +3404,7 @@ public boolean isWritable() { // or the pending publish bytes private void increasePendingSendRequestsAndPublishBytes(int msgSize) { if (++pendingSendRequest == maxPendingSendRequests) { - throttleTracker.setPendingSendRequestsExceeded(true); + throttleTracker.markThrottled(ThrottleType.ConnectionMaxQuantityOfInFlightPublishing); } PendingBytesPerThreadTracker.getInstance().incrementPublishBytes(msgSize, maxPendingBytesPerThread); } @@ -3424,7 +3429,7 @@ public void completedSendOperation(boolean isNonPersistentTopic, int msgSize) { PendingBytesPerThreadTracker.getInstance().decrementPublishBytes(msgSize, resumeThresholdPendingBytesPerThread); if (--pendingSendRequest == resumeReadsThreshold) { - throttleTracker.setPendingSendRequestsExceeded(false); + throttleTracker.unmarkThrottled(ThrottleType.ConnectionMaxQuantityOfInFlightPublishing); } if (isNonPersistentTopic) { @@ -3803,22 +3808,6 @@ protected void setAuthRole(String authRole) { this.authRole = authRole; } - /** - * {@inheritDoc} - */ - @Override - public void incrementThrottleCount() { - throttleTracker.incrementThrottleCount(); - } - - /** - * {@inheritDoc} - */ - @Override - public void decrementThrottleCount() { - throttleTracker.decrementThrottleCount(); - } - @VisibleForTesting void setAuthState(AuthenticationState authState) { this.authState = authState; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java index 78bac024218d8..45fa9aa2d7dc2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java @@ -18,125 +18,439 @@ */ package org.apache.pulsar.broker.service; -import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.broker.ServiceConfiguration; /** - * Tracks the state of throttling for a connection. The throttling happens by pausing reads by setting - * Netty {@link io.netty.channel.ChannelConfig#setAutoRead(boolean)} to false for the channel (connection). - *

- * There can be multiple rate limiters that can throttle a connection. Each rate limiter will independently - * call the {@link #incrementThrottleCount()} and {@link #decrementThrottleCount()} methods to signal that the - * connection should be throttled or not. The connection will be throttled if the counter is greater than 0. - *

- * Besides the rate limiters, the connection can also be throttled if the number of pending publish requests exceeds - * a configured threshold. This throttling is toggled with the {@link #setPendingSendRequestsExceeded} method. - * There's also per-thread memory limits which could throttle the connection. This throttling is toggled with the - * {@link #setPublishBufferLimiting} method. Internally, these two methods will call the - * {@link #incrementThrottleCount()} and {@link #decrementThrottleCount()} methods when the state changes. + * Manages and tracks throttling state for server connections in Apache Pulsar. + * + *

This class provides a centralized mechanism to control connection throttling by managing + * multiple throttling conditions simultaneously. When throttling is active, it pauses incoming + * requests by setting Netty's {@link io.netty.channel.ChannelConfig#setAutoRead(boolean)} to + * {@code false} for the associated channel. + * + *

Throttling Mechanism

+ *

The tracker maintains independent counters for different types of throttling conditions + * defined in {@link ThrottleType}. A connection is considered throttled if any of these + * conditions are active (counter > 0). The connection will only resume normal operation + * when all throttling conditions are cleared. + * + *

Supported Throttling Types

+ *
    + *
  • Connection-level: Max pending publish requests, outbound buffer limits
    • + *
  • Thread-level: IO thread memory limits for in-flight publishing
    • + *
  • Topic-level: Topic publish rate limiting
    • + *
  • Resource Group-level: Resource group publish rate limiting
    • + *
  • Broker-level: Global broker publish rate limiting
    • + *
  • Flow Control: Channel writability and cooldown rate limiting
    • + *
+ * + *

Reentrant vs Non-Reentrant Types

+ *

Some throttling types support multiple concurrent activations (reentrant): + *

    + *
  • {@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share + * the same rate limiter which relates to the same topic
    • + *
  • {@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share + * the same rate limiter which relates to the same resource group
    • + *
+ *

Other types are non-reentrant and can only be activated once at a time. The reentrant types + * use counters to track how many producers are affected by the same shared rate limiter, while + * non-reentrant types use simple boolean states. + * + *

Thread Safety

+ *

This class is designed to be used from a single thread (the connection's IO thread) + * and is not thread-safe for concurrent access from multiple threads. + * + *

Usage Example

+ * 
{@code
+ * ServerCnxThrottleTracker tracker = new ServerCnxThrottleTracker(serverCnx);
+ *
+ * // Mark connection as throttled due to rate limiting
+ * tracker.markThrottled(ThrottleType.TopicPublishRate);
+ *
+ * // Later, when rate limiting condition is cleared
+ * tracker.unmarkThrottled(ThrottleType.TopicPublishRate);
+ * }
+ * + * @see ThrottleType + * @see ThrottleRes + * @see ServerCnx */ @Slf4j -final class ServerCnxThrottleTracker { - - private static final AtomicIntegerFieldUpdater THROTTLE_COUNT_UPDATER = - AtomicIntegerFieldUpdater.newUpdater( - ServerCnxThrottleTracker.class, "throttleCount"); - - private static final AtomicIntegerFieldUpdater - PENDING_SEND_REQUESTS_EXCEEDED_UPDATER = - AtomicIntegerFieldUpdater.newUpdater( - ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded"); - private static final AtomicIntegerFieldUpdater PUBLISH_BUFFER_LIMITING_UPDATER = - AtomicIntegerFieldUpdater.newUpdater( - ServerCnxThrottleTracker.class, "publishBufferLimiting"); +public final class ServerCnxThrottleTracker { + private final ServerCnx serverCnx; - private volatile int throttleCount; - private volatile int pendingSendRequestsExceeded; - private volatile int publishBufferLimiting; + private final int[] states = new int[ThrottleType.values().length]; + + /** + * Enumeration of different throttling conditions that can be applied to a server connection. + * + *

Each type represents a specific resource constraint or rate limiting condition + * that may require throttling the connection to maintain system stability and fairness. + * + *

Some types support reentrant behavior (can be activated multiple times concurrently), + * while others are non-reentrant (single activation only). + */ + public static enum ThrottleType { + /** + * Throttling due to excessive pending publish requests on the connection. + * + *

This throttling is activated when the number of in-flight publish requests + * exceeds the configured limit. It helps prevent memory exhaustion and ensures + * fair resource allocation across connections. + * + *

Type: Non-reentrant + *

Configuration: {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()} + */ + ConnectionMaxQuantityOfInFlightPublishing, + /** + * Throttling due to excessive memory usage by in-flight publish operations on the IO thread. + * + *

This throttling is activated when the total memory used by pending publish operations + * on a shared IO thread exceeds the configured limit. Multiple connections may share the + * same IO thread, so this limit applies across all connections on that thread. + * + *

Type: Non-reentrant + *

Configuration: {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()} + */ + IOThreadMaxBytesOfInFlightPublishing, - public ServerCnxThrottleTracker(ServerCnx serverCnx) { - this.serverCnx = serverCnx; + /** + * Throttling due to topic-level publish rate limiting. + * + *

This throttling is activated when publish operations exceed the configured + * rate limits for a specific topic. Multiple producers on the same topic may + * contribute to triggering this throttling condition. + * + *

Type: Reentrant (supports multiple concurrent activations) + *
Reason for reentrancy: Multiple producers may share the same rate limiter + * which relates to the same topic. Each producer can independently trigger throttling + * when the shared topic rate limiter becomes active, requiring a counter to track + * how many producers are affected by the same rate limiter. + * + *

Configuration: Topic-level publish rate policies + */ + TopicPublishRate, + + /** + * Throttling due to resource group-level publish rate limiting. + * + *

This throttling is activated when publish operations exceed the configured + * rate limits for a resource group. Resource groups allow fine-grained control + * over resource allocation across multiple topics and tenants. + * + *

Type: Reentrant (supports multiple concurrent activations) + *
Reason for reentrancy: Multiple producers may share the same rate limiter + * which relates to the same resource group. Each producer can independently trigger + * throttling when the shared resource group rate limiter becomes active, requiring + * a counter to track how many producers are affected by the same rate limiter. + * + *

Configuration: Resource group publish rate policies + */ + ResourceGroupPublishRate, + /** + * Throttling due to broker-level publish rate limiting. + * + *

This throttling is activated when publish operations exceed the global + * broker-level rate limits. This provides a safety mechanism to prevent + * the entire broker from being overwhelmed by publish traffic. + * + *

Type: Non-reentrant + *

Configuration: {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()} + * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()} + */ + BrokerPublishRate, + + /** + * Throttling due to channel outbound buffer being full. + * + *

This throttling is activated when the Netty channel's outbound buffer + * reaches its high water mark, indicating that the client cannot keep up + * with the rate of outgoing messages. This prevents memory exhaustion + * and provides backpressure to publishers. + * + *

Type: Non-reentrant + *

Reference: PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK + */ + ConnectionOutboundBufferFull, + + /** + * Throttling due to connection pause/resume cooldown rate limiting. + * + *

This throttling is activated during cooldown periods after a connection + * has been resumed from a throttled state. It prevents rapid oscillation + * between throttled and unthrottled states. + * + *

Type: Non-reentrant + */ + ConnectionPauseReceivingCooldownRateLimit + } + + /** + * Enumeration representing the result of a throttling state change operation. + * + *

This enum indicates what happened when a throttling condition was marked or unmarked, + * helping callers understand whether the overall connection state changed or if the + * operation was ignored. + */ + enum ThrottleRes { + /** + * The operation resulted in a change to the overall connection throttling state. + * + *

This occurs when: + *

    + *
  • The connection transitions from unthrottled to throttled (first throttle type activated)
    • + *
  • The connection transitions from throttled to unthrottled (last throttle type deactivated)
    • + *
+ * + *

When this result is returned, the connection's auto-read setting will be updated + * accordingly to pause or resume request processing. + */ + ConnectionStateChanged, + + /** + * The operation changed the state of the specific throttle type but did not affect + * the overall connection throttling state. + * + *

This occurs when: + *

    + *
  • A throttle type is activated, but the connection was already throttled by other types
    • + *
  • A throttle type is deactivated, but the connection remains throttled by other types
    • + *
  • A reentrant throttle type's counter is incremented or decremented
    • + *
+ */ + TypeStateChanged, + + /** + * The operation was dropped because it would violate the throttle type's constraints. + * + *

This occurs when: + *

    + *
  • Attempting to mark a non-reentrant throttle type that is already active
    • + *
  • Attempting to unmark a throttle type that is not currently active
    • + *
  • Attempting to unmark a reentrant throttle type with an invalid counter state
    • + *
+ */ + Dropped } /** - * See {@link Producer#incrementThrottleCount()} for documentation. + * Checks if the connection is currently throttled by any throttle type. + * + *

This method examines all throttle type states and returns {@code true} + * if any of them are active (counter > 0). + * + * @return {@code true} if any throttling condition is active, {@code false} otherwise */ - public void incrementThrottleCount() { - int currentThrottleCount = THROTTLE_COUNT_UPDATER.incrementAndGet(this); - if (currentThrottleCount == 1) { - changeAutoRead(false); + private boolean hasThrottled() { + for (int stat : states) { + if (stat > 0) { + return true; + } } + return false; } /** - * See {@link Producer#decrementThrottleCount()} for documentation. + * Returns the total count of active throttling conditions across all types. + * + *

This method sums up all the individual counters for each throttle type, + * providing a measure of the overall throttling pressure on the connection. + * For reentrant types, this includes the full counter value (not just 0 or 1). + * + * @return the total number of active throttling conditions */ - public void decrementThrottleCount() { - int currentThrottleCount = THROTTLE_COUNT_UPDATER.decrementAndGet(this); - if (currentThrottleCount == 0) { - changeAutoRead(true); + public int throttledCount() { + int i = 0; + for (int stat : states) { + i += stat; } + return i; } - private void changeAutoRead(boolean autoRead) { - if (isChannelActive()) { + /** + * Marks the connection as throttled for the specified throttle type. + * + *

This method activates throttling for the given type and may pause the connection's + * request processing if this is the first active throttling condition. For reentrant + * types ({@link ThrottleType#TopicPublishRate} and {@link ThrottleType#ResourceGroupPublishRate}), + * this increments the counter. For non-reentrant types, this sets the state to active. + * + *

If the connection transitions from unthrottled to throttled, this method will + * set the Netty channel's auto-read to {@code false}, effectively pausing incoming + * request processing. + * + *

Metrics are automatically recorded to track throttling events and connection state changes. + * + * @param type the type of throttling condition to activate + * @throws IllegalArgumentException if type is null + * + * @see #unmarkThrottled(ThrottleType) + * @see ThrottleType + */ + public void markThrottled(ThrottleType type) { + ThrottleRes res = doMarkThrottled(type); + recordMetricsAfterThrottling(type, res); + if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { if (log.isDebugEnabled()) { - log.debug("[{}] Setting auto read to {}", serverCnx.toString(), autoRead); + log.debug("[{}] Setting auto read to false", serverCnx.toString()); } - // change the auto read flag on the channel - serverCnx.ctx().channel().config().setAutoRead(autoRead); - } - // update the metrics that track throttling - if (autoRead) { - serverCnx.getBrokerService().recordConnectionResumed(); - } else if (isChannelActive()) { - serverCnx.increasePublishLimitedTimesForTopics(); - serverCnx.getBrokerService().recordConnectionPaused(); + serverCnx.ctx().channel().config().setAutoRead(false); } } - private boolean isChannelActive() { - return serverCnx.isActive() && serverCnx.ctx() != null && serverCnx.ctx().channel().isActive(); + /** + * Unmarks the connection as throttled for the specified throttle type. + * + *

This method deactivates throttling for the given type and may resume the connection's + * request processing if this was the last active throttling condition. For reentrant + * types ({@link ThrottleType#TopicPublishRate} and {@link ThrottleType#ResourceGroupPublishRate}), + * this decrements the counter. For non-reentrant types, this clears the active state. + * + *

If the connection transitions from throttled to unthrottled, this method will + * set the Netty channel's auto-read to {@code true}, effectively resuming incoming + * request processing. + * + *

Metrics are automatically recorded to track unthrottling events and connection state changes. + * + * @param type the type of throttling condition to deactivate + * @throws IllegalArgumentException if type is null + * + * @see #markThrottled(ThrottleType) + * @see ThrottleType + */ + public void unmarkThrottled(ThrottleType type) { + ThrottleRes res = doUnmarkThrottled(type); + recordMetricsAfterUnthrottling(type, res); + if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { + if (log.isDebugEnabled()) { + log.debug("[{}] Setting auto read to true", serverCnx.toString()); + } + serverCnx.ctx().channel().config().setAutoRead(true); + } } - public void setPublishBufferLimiting(boolean throttlingEnabled) { - changeThrottlingFlag(PUBLISH_BUFFER_LIMITING_UPDATER, throttlingEnabled); + /** + * Internal method to mark a throttle type as active without side effects. + * + *

This method updates the internal state for the specified throttle type + * and returns the result of the operation. It handles both reentrant and + * non-reentrant throttle types appropriately: + * + *

    + *
  • Reentrant types: Increment the counter
    • + *
  • Non-reentrant types: Set to active (1) if not already active
    • + *
+ * + * @param throttleType the type of throttling to mark as active + * @return the result of the marking operation + * @see ThrottleRes + */ + private ThrottleRes doMarkThrottled(ThrottleType throttleType) { + // Two reentrant type: "TopicPublishRate" and "ResourceGroupPublishRate". + boolean throttled = hasThrottled(); + int value = states[throttleType.ordinal()]; + switch (throttleType) { + case TopicPublishRate: {} + case ResourceGroupPublishRate: { + states[throttleType.ordinal()] = value + 1; + return throttled ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; + } + default: { + states[throttleType.ordinal()] = 1; + if (value != 0) { + return ThrottleRes.Dropped; + } + return throttled ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; + } + } } - public void setPendingSendRequestsExceeded(boolean throttlingEnabled) { - boolean changed = changeThrottlingFlag(PENDING_SEND_REQUESTS_EXCEEDED_UPDATER, throttlingEnabled); - if (changed) { - // update the metrics that track throttling due to pending send requests - if (throttlingEnabled) { - serverCnx.getBrokerService().recordConnectionThrottled(); - } else { - serverCnx.getBrokerService().recordConnectionUnthrottled(); + /** + * Internal method to unmark a throttle type as active without side effects. + * + *

This method updates the internal state for the specified throttle type + * and returns the result of the operation. It handles both reentrant and + * non-reentrant throttle types appropriately: + * + *

    + *
  • Reentrant types: Decrement the counter
    • + *
  • Non-reentrant types: Clear active state if currently active
    • + *
+ * + * @param throttleType the type of throttling to mark as inactive + * @return the result of the unmarking operation + * @see ThrottleRes + */ + private ThrottleRes doUnmarkThrottled(ThrottleType throttleType) { + int value = states[throttleType.ordinal()]; + switch (throttleType) { + case TopicPublishRate: {} + case ResourceGroupPublishRate: { + states[throttleType.ordinal()] = value - 1; + return hasThrottled() ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; + } + default: { + if (value != 1) { + return ThrottleRes.Dropped; + } + states[throttleType.ordinal()] = 0; + return hasThrottled() ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; } } } + + /** + * Records metrics after a throttling operation. + * + *

This method updates various broker metrics to track throttling events: + *

    + *
  • Connection-specific throttling metrics for in-flight publishing limits
    • + *
  • Connection pause metrics when the overall connection state changes
    • + *
  • Topic-level publish limiting counters
    • + *
+ * + * @param type the throttle type that was activated + * @param res the result of the throttling operation + */ + private void recordMetricsAfterThrottling(ThrottleType type, ThrottleRes res) { + if (type == ThrottleType.ConnectionMaxQuantityOfInFlightPublishing && res != ThrottleRes.Dropped) { + serverCnx.getBrokerService().recordConnectionThrottled(); + } + if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { + serverCnx.increasePublishLimitedTimesForTopics(); + serverCnx.getBrokerService().recordConnectionPaused(); + } + } - private boolean changeThrottlingFlag(AtomicIntegerFieldUpdater throttlingFlagFieldUpdater, - boolean throttlingEnabled) { - // don't change a throttling flag if the channel is not active - if (!isChannelActive()) { - return false; + /** + * Records metrics after an unthrottling operation. + * + *

This method updates various broker metrics to track unthrottling events: + *

    + *
  • Connection-specific unthrottling metrics for in-flight publishing limits
    • + *
  • Connection resume metrics when the overall connection state changes
    • + *
+ * + * @param type the throttle type that was deactivated + * @param res the result of the unthrottling operation + */ + private void recordMetricsAfterUnthrottling(ThrottleType type, ThrottleRes res) { + if (type == ThrottleType.ConnectionMaxQuantityOfInFlightPublishing && res != ThrottleRes.Dropped) { + serverCnx.getBrokerService().recordConnectionUnthrottled(); } - if (throttlingFlagFieldUpdater.compareAndSet(this, booleanToInt(!throttlingEnabled), - booleanToInt(throttlingEnabled))) { - if (throttlingEnabled) { - incrementThrottleCount(); - } else { - decrementThrottleCount(); - } - return true; - } else { - return false; + if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { + serverCnx.getBrokerService().recordConnectionResumed(); } } - private static int booleanToInt(boolean value) { - return value ? 1 : 0; + public ServerCnxThrottleTracker(ServerCnx serverCnx) { + this.serverCnx = serverCnx; + } + + private boolean isChannelActive() { + return serverCnx.isActive() && serverCnx.ctx() != null && serverCnx.ctx().channel().isActive(); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java index 63599f09eef2e..2c0b247a94b7a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java @@ -88,22 +88,9 @@ public interface TransportCnx { CompletableFuture> checkConnectionLiveness(); /** - * Increments the counter that controls the throttling of the connection by pausing reads. - * The connection will be throttled while the counter is greater than 0. - *

- * The caller is responsible for decrementing the counter by calling {@link #decrementThrottleCount()} when the - * connection should no longer be throttled. + * Get the throttle tracker for this connection. */ - void incrementThrottleCount(); - - /** - * Decrements the counter that controls the throttling of the connection by pausing reads. - * The connection will be throttled while the counter is greater than 0. - *

- * This method should be called when the connection should no longer be throttled. However, the caller should have - * previously called {@link #incrementThrottleCount()}. - */ - void decrementThrottleCount(); + ServerCnxThrottleTracker getThrottleTracker(); FeatureFlags getFeatures(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java index ec952a7ca7734..2a536e1280ec7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java @@ -18,10 +18,14 @@ */ package org.apache.pulsar.broker.service; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import io.netty.channel.EventLoopGroup; import org.apache.pulsar.broker.qos.AsyncTokenBucket; +import org.testng.Assert; import org.testng.annotations.Test; public class PublishRateLimiterDisableTest { @@ -29,9 +33,23 @@ public class PublishRateLimiterDisableTest { // GH issue #10603 @Test void shouldAlwaysAllowAcquire() { - PublishRateLimiter publishRateLimiter = new PublishRateLimiterImpl(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK); + PublishRateLimiter publishRateLimiter = new PublishRateLimiterImpl(AsyncTokenBucket.DEFAULT_SNAPSHOT_CLOCK, + producer -> { + producer.getCnx().getThrottleTracker().markThrottled( + ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + }, producer -> { + producer.getCnx().getThrottleTracker().unmarkThrottled( + ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + }); Producer producer = mock(Producer.class); + ServerCnx serverCnx = mock(ServerCnx.class); + doAnswer(a -> serverCnx).when(producer).getCnx(); + ServerCnxThrottleTracker throttleTracker = new ServerCnxThrottleTracker(serverCnx); + doAnswer(a -> throttleTracker).when(serverCnx).getThrottleTracker(); + when(producer.getCnx()).thenReturn(serverCnx); + BrokerService brokerService = mock(BrokerService.class); + when(serverCnx.getBrokerService()).thenReturn(brokerService); publishRateLimiter.handlePublishThrottling(producer, Integer.MAX_VALUE, Long.MAX_VALUE); - verify(producer, never()).incrementThrottleCount(); + Assert.assertEquals(throttleTracker.throttledCount(), 0); } } \ No newline at end of file diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java index 20c1ad0a4125e..07ab9c41dc595 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java @@ -19,18 +19,16 @@ package org.apache.pulsar.broker.service; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import io.netty.channel.DefaultEventLoop; import io.netty.channel.EventLoop; import io.netty.channel.EventLoopGroup; +import io.netty.util.concurrent.DefaultThreadFactory; import java.util.HashMap; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.PublishRate; @@ -47,36 +45,37 @@ public class PublishRateLimiterTest { private AtomicLong manualClockSource; private Producer producer; + private ServerCnx serverCnx; private PublishRateLimiterImpl publishRateLimiter; - - private AtomicInteger throttleCount = new AtomicInteger(0); + private ServerCnxThrottleTracker throttleTracker; + private DefaultThreadFactory threadFactory = new DefaultThreadFactory("pulsar-io"); + private EventLoop eventLoop = new DefaultEventLoop(threadFactory); @BeforeMethod public void setup() throws Exception { policies.publishMaxMessageRate = new HashMap<>(); policies.publishMaxMessageRate.put(CLUSTER_NAME, publishRate); manualClockSource = new AtomicLong(TimeUnit.SECONDS.toNanos(100)); - publishRateLimiter = new PublishRateLimiterImpl(() -> manualClockSource.get()); + publishRateLimiter = new PublishRateLimiterImpl(() -> manualClockSource.get(), + producer -> { + producer.getCnx().getThrottleTracker().markThrottled( + ServerCnxThrottleTracker.ThrottleType.TopicPublishRate); + }, producer -> { + producer.getCnx().getThrottleTracker().unmarkThrottled( + ServerCnxThrottleTracker.ThrottleType.TopicPublishRate); + }); publishRateLimiter.update(policies, CLUSTER_NAME); producer = mock(Producer.class); - throttleCount.set(0); - doAnswer(a -> { - throttleCount.incrementAndGet(); - return null; - }).when(producer).incrementThrottleCount(); - doAnswer(a -> { - throttleCount.decrementAndGet(); - return null; - }).when(producer).decrementThrottleCount(); - TransportCnx transportCnx = mock(TransportCnx.class); - when(producer.getCnx()).thenReturn(transportCnx); + serverCnx = mock(ServerCnx.class); + doAnswer(a -> this.serverCnx).when(producer).getCnx(); + throttleTracker = new ServerCnxThrottleTracker(this.serverCnx); + doAnswer(a -> throttleTracker).when(this.serverCnx).getThrottleTracker(); + when(producer.getCnx()).thenReturn(serverCnx); BrokerService brokerService = mock(BrokerService.class); - when(transportCnx.getBrokerService()).thenReturn(brokerService); + when(serverCnx.getBrokerService()).thenReturn(brokerService); EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class); when(brokerService.executor()).thenReturn(eventLoopGroup); - EventLoop eventLoop = mock(EventLoop.class); when(eventLoopGroup.next()).thenReturn(eventLoop); - doReturn(null).when(eventLoop).schedule(any(Runnable.class), anyLong(), any()); incrementSeconds(1); } @@ -86,6 +85,11 @@ public void cleanup() throws Exception { policies.publishMaxMessageRate = null; } + @AfterMethod + public void tearDown() throws Exception { + eventLoop.shutdownGracefully(); + } + private void incrementSeconds(int seconds) { manualClockSource.addAndGet(TimeUnit.SECONDS.toNanos(seconds)); } @@ -94,30 +98,30 @@ private void incrementSeconds(int seconds) { public void testPublishRateLimiterImplExceed() throws Exception { // increment not exceed publishRateLimiter.handlePublishThrottling(producer, 5, 50); - assertEquals(throttleCount.get(), 0); + assertEquals(throttleTracker.throttledCount(), 0); incrementSeconds(1); // numOfMessages increment exceeded publishRateLimiter.handlePublishThrottling(producer, 11, 100); - assertEquals(throttleCount.get(), 1); + assertEquals(throttleTracker.throttledCount(), 1); incrementSeconds(1); // msgSizeInBytes increment exceeded publishRateLimiter.handlePublishThrottling(producer, 9, 110); - assertEquals(throttleCount.get(), 2); + assertEquals(throttleTracker.throttledCount(), 2); } @Test public void testPublishRateLimiterImplUpdate() { publishRateLimiter.handlePublishThrottling(producer, 11, 110); - assertEquals(throttleCount.get(), 1); + assertEquals(throttleTracker.throttledCount(), 1); // update - throttleCount.set(0); + throttleTracker = new ServerCnxThrottleTracker(serverCnx); publishRateLimiter.update(newPublishRate); publishRateLimiter.handlePublishThrottling(producer, 11, 110); - assertEquals(throttleCount.get(), 0); + assertEquals(throttleTracker.throttledCount(), 0); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java index 40bcb19ab0ca5..929350b599e76 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPublishRateThrottleTest.java @@ -73,6 +73,37 @@ public void testProducerBlockedByPrecisTopicPublishRateLimiting() throws Excepti pulsarClient.close(); } + @Test + public void testResumeEvenProducerClosed() throws Exception { + PublishRate publishRate = new PublishRate(1, 10); + conf.setMaxPendingPublishRequestsPerConnection(0); + super.baseSetup(); + admin.namespaces().setPublishRate("prop/ns-abc", publishRate); + final String topic = BrokerTestUtil.newUniqueName("persistent://prop/ns-abc/tp"); + org.apache.pulsar.client.api.Producer producer = pulsarClient.newProducer() + .topic(topic).create(); + + Topic topicRef = pulsar.getBrokerService().getTopicReference(topic).get(); + Assert.assertNotNull(topicRef); + MessageId messageId = null; + // first will be success, and the second will fail, will set auto read to false. + messageId = producer.sendAsync(new byte[10]).get(500, TimeUnit.MILLISECONDS); + Assert.assertNotNull(messageId); + // second will be blocked + producer.sendAsync(new byte[10]); + + // Verify: even through the producer was closed before the unblock, the state should be unblocked at the next + // period of rate limiter. + producer.close(); + Thread.sleep(3000); + org.apache.pulsar.client.api.Producer producer2 = pulsarClient.newProducer() + .topic(topic).create(); + producer2.sendAsync(new byte[2]).get(500, TimeUnit.MILLISECONDS); + + // Close the PulsarClient gracefully to avoid ByteBuf leak + pulsarClient.close(); + } + @Test public void testSystemTopicPublishNonBlock() throws Exception { super.baseSetup(); From 2070daaade7c630d3ab1062af8fc378f6ae8db27 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 9 Oct 2025 11:01:08 +0800 Subject: [PATCH 2/8] add assert: eusure all throttle status changing is in the same thread --- .../pulsar/broker/service/ServerCnxThrottleTracker.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java index 45fa9aa2d7dc2..da750f96647d7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.service; +import com.google.common.annotations.VisibleForTesting; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.ServiceConfiguration; @@ -262,6 +263,7 @@ private boolean hasThrottled() { * * @return the total number of active throttling conditions */ + @VisibleForTesting public int throttledCount() { int i = 0; for (int stat : states) { @@ -291,6 +293,7 @@ public int throttledCount() { * @see ThrottleType */ public void markThrottled(ThrottleType type) { + assert serverCnx.ctx().executor().inEventLoop() : "This method should be called in serverCnx.ctx().executor()"; ThrottleRes res = doMarkThrottled(type); recordMetricsAfterThrottling(type, res); if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { @@ -322,6 +325,7 @@ public void markThrottled(ThrottleType type) { * @see ThrottleType */ public void unmarkThrottled(ThrottleType type) { + assert serverCnx.ctx().executor().inEventLoop() : "This method should be called in serverCnx.ctx().executor()"; ThrottleRes res = doUnmarkThrottled(type); recordMetricsAfterUnthrottling(type, res); if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { From f37428d68b43c2f7c6d7a39b23e70e039666b536 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 9 Oct 2025 11:26:35 +0800 Subject: [PATCH 3/8] add new field reentrant --- .../ResourceGroupPublishLimiter.java | 4 +- .../service/ServerCnxThrottleTracker.java | 61 +++++++++---------- 2 files changed, 32 insertions(+), 33 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java index edc723c35e94c..04f56e0ca69ac 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java @@ -33,10 +33,10 @@ public class ResourceGroupPublishLimiter extends PublishRateLimiterImpl { public ResourceGroupPublishLimiter(ResourceGroup resourceGroup, MonotonicClock monotonicClock) { super(monotonicClock, producer -> { producer.getCnx().getThrottleTracker().markThrottled( - ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + ServerCnxThrottleTracker.ThrottleType.ResourceGroupPublishRate); }, producer -> { producer.getCnx().getThrottleTracker().unmarkThrottled( - ServerCnxThrottleTracker.ThrottleType.BrokerPublishRate); + ServerCnxThrottleTracker.ThrottleType.ResourceGroupPublishRate); }); update(resourceGroup); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java index da750f96647d7..d05d530086aac 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java @@ -19,6 +19,7 @@ package org.apache.pulsar.broker.service; import com.google.common.annotations.VisibleForTesting; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.ServiceConfiguration; @@ -93,6 +94,7 @@ public final class ServerCnxThrottleTracker { * while others are non-reentrant (single activation only). */ public static enum ThrottleType { + /** * Throttling due to excessive pending publish requests on the connection. * @@ -103,7 +105,7 @@ public static enum ThrottleType { *

Type: Non-reentrant *

Configuration: {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()} */ - ConnectionMaxQuantityOfInFlightPublishing, + ConnectionMaxQuantityOfInFlightPublishing(false), /** * Throttling due to excessive memory usage by in-flight publish operations on the IO thread. @@ -115,7 +117,7 @@ public static enum ThrottleType { *

Type: Non-reentrant *

Configuration: {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()} */ - IOThreadMaxBytesOfInFlightPublishing, + IOThreadMaxBytesOfInFlightPublishing(false), /** * Throttling due to topic-level publish rate limiting. @@ -132,7 +134,7 @@ public static enum ThrottleType { * *

Configuration: Topic-level publish rate policies */ - TopicPublishRate, + TopicPublishRate(true), /** * Throttling due to resource group-level publish rate limiting. @@ -149,7 +151,7 @@ public static enum ThrottleType { * *

Configuration: Resource group publish rate policies */ - ResourceGroupPublishRate, + ResourceGroupPublishRate(true), /** * Throttling due to broker-level publish rate limiting. @@ -162,7 +164,7 @@ public static enum ThrottleType { *

Configuration: {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()} * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()} */ - BrokerPublishRate, + BrokerPublishRate(false), /** * Throttling due to channel outbound buffer being full. @@ -175,7 +177,7 @@ public static enum ThrottleType { *

Type: Non-reentrant *

Reference: PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK */ - ConnectionOutboundBufferFull, + ConnectionOutboundBufferFull(false), /** * Throttling due to connection pause/resume cooldown rate limiting. @@ -186,7 +188,14 @@ public static enum ThrottleType { * *

Type: Non-reentrant */ - ConnectionPauseReceivingCooldownRateLimit + ConnectionPauseReceivingCooldownRateLimit(false); + + @Getter + final boolean reentrant; + + ThrottleType(boolean reentrant) { + this.reentrant = reentrant; + } } /** @@ -356,20 +365,15 @@ private ThrottleRes doMarkThrottled(ThrottleType throttleType) { // Two reentrant type: "TopicPublishRate" and "ResourceGroupPublishRate". boolean throttled = hasThrottled(); int value = states[throttleType.ordinal()]; - switch (throttleType) { - case TopicPublishRate: {} - case ResourceGroupPublishRate: { - states[throttleType.ordinal()] = value + 1; - return throttled ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; - } - default: { - states[throttleType.ordinal()] = 1; - if (value != 0) { - return ThrottleRes.Dropped; - } - return throttled ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; + if (throttleType.isReentrant()) { + states[throttleType.ordinal()] = value + 1; + } else { + states[throttleType.ordinal()] = 1; + if (value != 0) { + return ThrottleRes.Dropped; } } + return throttled ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; } /** @@ -390,20 +394,15 @@ private ThrottleRes doMarkThrottled(ThrottleType throttleType) { */ private ThrottleRes doUnmarkThrottled(ThrottleType throttleType) { int value = states[throttleType.ordinal()]; - switch (throttleType) { - case TopicPublishRate: {} - case ResourceGroupPublishRate: { - states[throttleType.ordinal()] = value - 1; - return hasThrottled() ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; - } - default: { - if (value != 1) { - return ThrottleRes.Dropped; - } - states[throttleType.ordinal()] = 0; - return hasThrottled() ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; + if (throttleType.isReentrant()) { + states[throttleType.ordinal()] = value - 1; + } else { + if (value != 1) { + return ThrottleRes.Dropped; } + states[throttleType.ordinal()] = 0; } + return hasThrottled() ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; } /** From cb47b19e5b2b9630b2908c2edcac4b82a7cf485c Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 14 Oct 2025 22:00:51 +0800 Subject: [PATCH 4/8] improve enum names --- .../java/org/apache/pulsar/broker/service/ServerCnx.java | 8 ++++---- .../pulsar/broker/service/ServerCnxThrottleTracker.java | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 25642a0b670b1..adbd82e80cedb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -285,7 +285,7 @@ public void incrementPublishBytes(long bytes, long maxPendingBytesPerThread) { && !limitExceeded) { limitExceeded = true; cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.markThrottled( - ThrottleType.IOThreadMaxBytesOfInFlightPublishing)); + ThrottleType.IOThreadMaxPendingPublishBytesExceeded)); } } @@ -296,7 +296,7 @@ public void decrementPublishBytes(long bytes, long resumeThresholdPendingBytesPe if (limitExceeded && pendingBytes <= resumeThresholdPendingBytesPerThread) { limitExceeded = false; cnxsPerThread.get().forEach(cnx -> cnx.throttleTracker.unmarkThrottled( - ThrottleType.IOThreadMaxBytesOfInFlightPublishing)); + ThrottleType.IOThreadMaxPendingPublishBytesExceeded)); } } } @@ -3404,7 +3404,7 @@ public boolean isWritable() { // or the pending publish bytes private void increasePendingSendRequestsAndPublishBytes(int msgSize) { if (++pendingSendRequest == maxPendingSendRequests) { - throttleTracker.markThrottled(ThrottleType.ConnectionMaxQuantityOfInFlightPublishing); + throttleTracker.markThrottled(ThrottleType.ConnectionMaxPendingPublishRequestsExceeded); } PendingBytesPerThreadTracker.getInstance().incrementPublishBytes(msgSize, maxPendingBytesPerThread); } @@ -3429,7 +3429,7 @@ public void completedSendOperation(boolean isNonPersistentTopic, int msgSize) { PendingBytesPerThreadTracker.getInstance().decrementPublishBytes(msgSize, resumeThresholdPendingBytesPerThread); if (--pendingSendRequest == resumeReadsThreshold) { - throttleTracker.unmarkThrottled(ThrottleType.ConnectionMaxQuantityOfInFlightPublishing); + throttleTracker.unmarkThrottled(ThrottleType.ConnectionMaxPendingPublishRequestsExceeded); } if (isNonPersistentTopic) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java index d05d530086aac..d705ed7d591b4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java @@ -105,7 +105,7 @@ public static enum ThrottleType { *

Type: Non-reentrant *

Configuration: {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()} */ - ConnectionMaxQuantityOfInFlightPublishing(false), + ConnectionMaxPendingPublishRequestsExceeded(false), /** * Throttling due to excessive memory usage by in-flight publish operations on the IO thread. @@ -117,7 +117,7 @@ public static enum ThrottleType { *

Type: Non-reentrant *

Configuration: {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()} */ - IOThreadMaxBytesOfInFlightPublishing(false), + IOThreadMaxPendingPublishBytesExceeded(false), /** * Throttling due to topic-level publish rate limiting. @@ -419,7 +419,7 @@ private ThrottleRes doUnmarkThrottled(ThrottleType throttleType) { * @param res the result of the throttling operation */ private void recordMetricsAfterThrottling(ThrottleType type, ThrottleRes res) { - if (type == ThrottleType.ConnectionMaxQuantityOfInFlightPublishing && res != ThrottleRes.Dropped) { + if (type == ThrottleType.ConnectionMaxPendingPublishRequestsExceeded && res != ThrottleRes.Dropped) { serverCnx.getBrokerService().recordConnectionThrottled(); } if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { @@ -441,7 +441,7 @@ private void recordMetricsAfterThrottling(ThrottleType type, ThrottleRes res) { * @param res the result of the unthrottling operation */ private void recordMetricsAfterUnthrottling(ThrottleType type, ThrottleRes res) { - if (type == ThrottleType.ConnectionMaxQuantityOfInFlightPublishing && res != ThrottleRes.Dropped) { + if (type == ThrottleType.ConnectionMaxPendingPublishRequestsExceeded && res != ThrottleRes.Dropped) { serverCnx.getBrokerService().recordConnectionUnthrottled(); } if (res == ThrottleRes.ConnectionStateChanged && isChannelActive()) { From f49a7f6c33aaf95c6c6074bef82c58c38d2039d9 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 15 Oct 2025 15:09:25 +0800 Subject: [PATCH 5/8] checkstyle --- .../apache/pulsar/broker/service/AbstractTopic.java | 11 ++++++----- .../org/apache/pulsar/broker/service/ServerCnx.java | 2 +- .../broker/service/ServerCnxThrottleTracker.java | 4 ++-- .../broker/service/PublishRateLimiterDisableTest.java | 3 --- 4 files changed, 9 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java index f1b0d5eadfe82..24bce1e39bb8b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java @@ -194,11 +194,12 @@ public AbstractTopic(String topic, BrokerService brokerService) { updateTopicPolicyByBrokerConfig(); this.lastActive = System.nanoTime(); - topicPublishRateLimiter = new PublishRateLimiterImpl(brokerService.getPulsar().getMonotonicClock(), producer -> { - producer.getCnx().getThrottleTracker().markThrottled(ThrottleType.TopicPublishRate); - }, producer -> { - producer.getCnx().getThrottleTracker().unmarkThrottled(ThrottleType.TopicPublishRate); - }); + topicPublishRateLimiter = new PublishRateLimiterImpl(brokerService.getPulsar().getMonotonicClock(), + producer -> { + producer.getCnx().getThrottleTracker().markThrottled(ThrottleType.TopicPublishRate); + }, producer -> { + producer.getCnx().getThrottleTracker().unmarkThrottled(ThrottleType.TopicPublishRate); + }); updateActiveRateLimiters(); additionalSystemCursorNames = brokerService.pulsar().getConfiguration().getAdditionalSystemCursorNames(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index adbd82e80cedb..b11869e06b5d2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -26,8 +26,8 @@ import static org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.unsafeGetPartitionedTopicMetadataAsync; import static org.apache.pulsar.broker.lookup.TopicLookupBase.lookupTopicAsync; import static org.apache.pulsar.broker.service.persistent.PersistentTopic.getMigratedClusterUrl; -import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.ignoreUnrecoverableBKException; import static org.apache.pulsar.broker.service.ServerCnxThrottleTracker.ThrottleType; +import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.ignoreUnrecoverableBKException; import static org.apache.pulsar.common.api.proto.ProtocolVersion.v5; import static org.apache.pulsar.common.naming.Constants.WEBSOCKET_DUMMY_ORIGINAL_PRINCIPLE; import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java index d705ed7d591b4..037612d21564b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java @@ -93,7 +93,7 @@ public final class ServerCnxThrottleTracker { *

Some types support reentrant behavior (can be activated multiple times concurrently), * while others are non-reentrant (single activation only). */ - public static enum ThrottleType { + public enum ThrottleType { /** * Throttling due to excessive pending publish requests on the connection. @@ -404,7 +404,7 @@ private ThrottleRes doUnmarkThrottled(ThrottleType throttleType) { } return hasThrottled() ? ThrottleRes.TypeStateChanged : ThrottleRes.ConnectionStateChanged; } - + /** * Records metrics after a throttling operation. * diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java index 2a536e1280ec7..3e6edb04932eb 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterDisableTest.java @@ -20,10 +20,7 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -import io.netty.channel.EventLoopGroup; import org.apache.pulsar.broker.qos.AsyncTokenBucket; import org.testng.Assert; import org.testng.annotations.Test; From 39bfe9e8065b8588e014162e1c01e642ef586e99 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 15 Oct 2025 20:31:05 +0800 Subject: [PATCH 6/8] fix test --- .../service/PublishRateLimiterTest.java | 65 +++++++++++++------ 1 file changed, 45 insertions(+), 20 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java index 07ab9c41dc595..573e3980c7383 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java @@ -23,11 +23,13 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import io.netty.channel.ChannelHandlerContext; import io.netty.channel.DefaultEventLoop; import io.netty.channel.EventLoop; import io.netty.channel.EventLoopGroup; import io.netty.util.concurrent.DefaultThreadFactory; import java.util.HashMap; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import org.apache.pulsar.common.policies.data.Policies; @@ -67,6 +69,9 @@ public void setup() throws Exception { publishRateLimiter.update(policies, CLUSTER_NAME); producer = mock(Producer.class); serverCnx = mock(ServerCnx.class); + ChannelHandlerContext channelHandlerContext = mock(ChannelHandlerContext.class); + doAnswer(a -> eventLoop).when(channelHandlerContext).executor(); + doAnswer(a -> channelHandlerContext).when(serverCnx).ctx(); doAnswer(a -> this.serverCnx).when(producer).getCnx(); throttleTracker = new ServerCnxThrottleTracker(this.serverCnx); doAnswer(a -> throttleTracker).when(this.serverCnx).getThrottleTracker(); @@ -96,32 +101,52 @@ private void incrementSeconds(int seconds) { @Test public void testPublishRateLimiterImplExceed() throws Exception { - // increment not exceed - publishRateLimiter.handlePublishThrottling(producer, 5, 50); - assertEquals(throttleTracker.throttledCount(), 0); + CompletableFuture future = new CompletableFuture<>(); + eventLoop.execute(() -> { + try { + // increment not exceed + publishRateLimiter.handlePublishThrottling(producer, 5, 50); + assertEquals(throttleTracker.throttledCount(), 0); - incrementSeconds(1); + incrementSeconds(1); - // numOfMessages increment exceeded - publishRateLimiter.handlePublishThrottling(producer, 11, 100); - assertEquals(throttleTracker.throttledCount(), 1); + // numOfMessages increment exceeded + publishRateLimiter.handlePublishThrottling(producer, 11, 100); + assertEquals(throttleTracker.throttledCount(), 1); - incrementSeconds(1); + incrementSeconds(1); - // msgSizeInBytes increment exceeded - publishRateLimiter.handlePublishThrottling(producer, 9, 110); - assertEquals(throttleTracker.throttledCount(), 2); + // msgSizeInBytes increment exceeded + publishRateLimiter.handlePublishThrottling(producer, 9, 110); + assertEquals(throttleTracker.throttledCount(), 2); + + future.complete(null); + } catch (Exception e) { + future.completeExceptionally(e); + } + }); + future.get(5, TimeUnit.SECONDS); } @Test - public void testPublishRateLimiterImplUpdate() { - publishRateLimiter.handlePublishThrottling(producer, 11, 110); - assertEquals(throttleTracker.throttledCount(), 1); - - // update - throttleTracker = new ServerCnxThrottleTracker(serverCnx); - publishRateLimiter.update(newPublishRate); - publishRateLimiter.handlePublishThrottling(producer, 11, 110); - assertEquals(throttleTracker.throttledCount(), 0); + public void testPublishRateLimiterImplUpdate() throws Exception { + CompletableFuture future = new CompletableFuture<>(); + eventLoop.execute(() -> { + try { + publishRateLimiter.handlePublishThrottling(producer, 11, 110); + assertEquals(throttleTracker.throttledCount(), 1); + + // update + throttleTracker = new ServerCnxThrottleTracker(serverCnx); + publishRateLimiter.update(newPublishRate); + publishRateLimiter.handlePublishThrottling(producer, 11, 110); + assertEquals(throttleTracker.throttledCount(), 0); + + future.complete(null); + } catch (Exception e) { + future.completeExceptionally(e); + } + }); + future.get(5, TimeUnit.SECONDS); } } From 3d2b4a327f8a905b49352a53d236d0d01008fdc1 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 15 Oct 2025 20:33:54 +0800 Subject: [PATCH 7/8] checkstyle --- .../main/java/org/apache/pulsar/broker/service/ServerCnx.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index b11869e06b5d2..5497f35e4de4d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -25,8 +25,8 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank; import static org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.unsafeGetPartitionedTopicMetadataAsync; import static org.apache.pulsar.broker.lookup.TopicLookupBase.lookupTopicAsync; -import static org.apache.pulsar.broker.service.persistent.PersistentTopic.getMigratedClusterUrl; import static org.apache.pulsar.broker.service.ServerCnxThrottleTracker.ThrottleType; +import static org.apache.pulsar.broker.service.persistent.PersistentTopic.getMigratedClusterUrl; import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.ignoreUnrecoverableBKException; import static org.apache.pulsar.common.api.proto.ProtocolVersion.v5; import static org.apache.pulsar.common.naming.Constants.WEBSOCKET_DUMMY_ORIGINAL_PRINCIPLE; From 22213326ab999f49c1a41c8791b43364f26f3a28 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 15 Oct 2025 20:51:43 +0800 Subject: [PATCH 8/8] solve issue --- .../java/org/apache/pulsar/broker/service/ServerCnx.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 5497f35e4de4d..d7010e3cf8c7c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -501,7 +501,7 @@ private void checkPauseReceivingRequestsAfterResumeRateLimit(BaseCommand cmd) { public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { if (pauseReceivingRequestsIfUnwritable && ctx.channel().isWritable()) { log.info("[{}] is writable, turn on channel auto-read", this); - getThrottleTracker().markThrottled(ThrottleType.ConnectionOutboundBufferFull); + getThrottleTracker().unmarkThrottled(ThrottleType.ConnectionOutboundBufferFull); requestRateLimiter.timingOpen(pauseReceivingCooldownMilliSeconds, TimeUnit.MILLISECONDS); } else if (pauseReceivingRequestsIfUnwritable && !ctx.channel().isWritable()) { final ChannelOutboundBuffer outboundBuffer = ctx.channel().unsafe().outboundBuffer(); @@ -515,8 +515,7 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exceptio PAUSE_RECEIVING_LOG.debug("[{}] is not writable, turn off channel auto-read", this); } } - getThrottleTracker().unmarkThrottled(ThrottleType.ConnectionOutboundBufferFull); - ctx.channel().config().setAutoRead(false); + getThrottleTracker().markThrottled(ThrottleType.ConnectionOutboundBufferFull); } ctx.fireChannelWritabilityChanged(); }