You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2025/09/29 06:36:43 UTC
[PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
poorbarcode opened a new pull request, #24799:
URL: https://github.com/apache/pulsar/pull/24799
### Motivation
See the PIP-434
- https://github.com/apache/pulsar/pull/24510
### Modifications
- Use `ServerCnxThrottleTracker`, instead of modifying `channel.readable` directly
- Changes of `ServerCnxThrottleTracker`
- Merge multiple tracker states into one.
- Split the counter for each throttling type, to avoid them affecting each other, and make troubleshooting easier.
- Instead of modifying per tracker state with CAS, let all changes in the same thread(`pulsar-io-thread`)
- Add a new test to confirm that the limited state will be resumed after a period of the rate limiter, even though the producer was closed
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: x
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2415469581
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ * <li><b>Reentrant types:</b> Increment the counter</li>
+ * <li><b>Non-reentrant types:</b> Set to active (1) if not already active</li>
+ * </ul>
+ *
+ * @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;
Review Comment:
> it would be useful to log this case just so that the ThrottleTypes aren't misused later.
I added a boolean field named `reentrant` for the `ThrottleType`, to manually define a value for the new field is needed when a new type is added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2443890340
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
Review Comment:
For now, it's okay to keep this change. Because if calling `setAutoRead` out of the Netty thread does not work, the previous publish rate limiting feature might actually never work as expected.
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
Review Comment:
For now, it's okay to keep this change. Because if calling `setAutoRead` out of the Netty thread does not work, the previous publish rate limiting feature in KoP might actually never work as expected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#issuecomment-3406192080
Rebase master
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2415473586
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ * <li><b>Reentrant types:</b> Increment the counter</li>
+ * <li><b>Non-reentrant types:</b> Set to active (1) if not already active</li>
+ * </ul>
+ *
+ * @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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ * <li><b>Reentrant types:</b> Decrement the counter</li>
+ * <li><b>Non-reentrant types:</b> Clear active state if currently active</li>
+ * </ul>
+ *
+ * @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;
Review Comment:
Improved. See also https://github.com/apache/pulsar/pull/24799#discussion_r2415469581
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2388772312
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ * <li><b>Reentrant types:</b> Increment the counter</li>
+ * <li><b>Non-reentrant types:</b> Set to active (1) if not already active</li>
+ * </ul>
+ *
+ * @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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ * <li><b>Reentrant types:</b> Decrement the counter</li>
+ * <li><b>Non-reentrant types:</b> Clear active state if currently active</li>
+ * </ul>
+ *
+ * @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;
Review Comment:
log this case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2388746285
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
Review Comment:
At the beginning of the method, could we add a check here to ensure that this is executed within the event loop of the connection?
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
Review Comment:
At the beginning of the method, could we add a check here to ensure that this is executed within the event loop of the connection?
something like
```
assert serverCnx.ctx().executor().inEventLoop() : "This method should be called in serverCnx.ctx().executor()";
```
Another possibility would be to always accept the call and if it's not in the eventloop, then run the call to the event loop.
```
if (!serverCnx.ctx().executor().inEventLoop()) {
serverCnx.ctx().executor().execute(() -> markThrottled(type));
}
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java:
##########
@@ -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);
Review Comment:
should use ResourceGroupPublishRate here?
just wondering if the ResourceGroup solution allows multiple ResourceGroupPublishLimiters being active at once. In that case a single ResourceGroupPublishRate wouldn't be sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#issuecomment-3404565985
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2428087569
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java:
##########
@@ -18,125 +18,442 @@
*/
package org.apache.pulsar.broker.service;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import com.google.common.annotations.VisibleForTesting;
+import lombok.Getter;
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>Some types support reentrant behavior (can be activated multiple times concurrently),
+ * while others are non-reentrant (single activation only).
+ */
+ public static enum ThrottleType {
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to excessive pending publish requests on the connection.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing(false),
Review Comment:
the name `ConnectionMaxQuantityOfInFlightPublishing` is not great. It's better to align more closely with `maxPendingPublishRequestsPerConnection` configuration setting.
For example, `ConnectionMaxPendingPublishRequestsExceeded` (could also omit `Exceeded` suffix).
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java:
##########
@@ -18,125 +18,442 @@
*/
package org.apache.pulsar.broker.service;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import com.google.common.annotations.VisibleForTesting;
+import lombok.Getter;
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>Some types support reentrant behavior (can be activated multiple times concurrently),
+ * while others are non-reentrant (single activation only).
+ */
+ public static enum ThrottleType {
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to excessive pending publish requests on the connection.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing(false),
+
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing(false),
Review Comment:
The name `IOThreadMaxBytesOfInFlightPublishing` isn't great. It's better to more closely alight with the configuration setting `maxMessagePublishBufferSizeInMB`.
For example `IOThreadMaxPendingPublishBytesExceeded`. This would also be aligned with the possible new name `ConnectionMaxPendingPublishRequestsExceeded`.
(could also omit `Exceeded` suffixes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2428091890
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnxThrottleTracker.java:
##########
@@ -18,125 +18,442 @@
*/
package org.apache.pulsar.broker.service;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import com.google.common.annotations.VisibleForTesting;
+import lombok.Getter;
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>Some types support reentrant behavior (can be activated multiple times concurrently),
+ * while others are non-reentrant (single activation only).
+ */
+ public static enum ThrottleType {
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to excessive pending publish requests on the connection.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing(false),
+
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing(false),
Review Comment:
The name `IOThreadMaxBytesOfInFlightPublishing` isn't great. It's better to more closely aligned with the configuration setting `maxMessagePublishBufferSizeInMB`.
For example `IOThreadMaxPendingPublishBytesExceeded`. This would also be aligned with the possible new name `ConnectionMaxPendingPublishRequestsExceeded`.
(could also omit `Exceeded` suffixes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2415448600
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
Review Comment:
Added the check, see also https://github.com/apache/pulsar/pull/24799#discussion_r2415448306
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2415453063
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java:
##########
@@ -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);
Review Comment:
> should use ResourceGroupPublishRate here?
Thanks for mentioning this mistake. I have changed it
> just wondering if the ResourceGroup solution allows multiple ResourceGroupPublishLimiters being active at once. In that case a single ResourceGroupPublishRate wouldn't be sufficient.
Since the `markThrottled` is reentrant if the type is `ResourceGroupPublishRate`, it should work. Could you explain more details?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2388771533
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ * <li><b>Reentrant types:</b> Increment the counter</li>
+ * <li><b>Non-reentrant types:</b> Set to active (1) if not already active</li>
+ * </ul>
+ *
+ * @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;
Review Comment:
it would be useful to log this case just so that the ThrottleTypes aren't misused later. Let's say if there would be a case in the future where a since ThrottleType has multiple instances behind it. That should be handled by adding a counter for ThrottleType. I guess we don't have that currently unless the ResourceGroupPublishType turns out to be one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2428103657
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupPublishLimiter.java:
##########
@@ -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);
Review Comment:
This looks fine now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2415448306
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
Review Comment:
Sure, added the check for both `markThrottled` and `unmarkThrottled`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#issuecomment-3406896252
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24799?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `85.10638%` with `14 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.30%. Comparing base ([`a091ea7`](https://app.codecov.io/gh/apache/pulsar/commit/a091ea74de8b60f00fdb380584c7d49279610ca1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`2221332`](https://app.codecov.io/gh/apache/pulsar/commit/22213326ab999f49c1a41c8791b43364f26f3a28?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24799?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ulsar/broker/service/ServerCnxThrottleTracker.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FServerCnxThrottleTracker.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1NlcnZlckNueFRocm90dGxlVHJhY2tlci5qYXZh) | 81.53% | [3 Missing and 9 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...va/org/apache/pulsar/broker/service/ServerCnx.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FServerCnx.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1NlcnZlckNueC5qYXZh) | 75.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24799 +/- ##
============================================
+ Coverage 74.29% 74.30% +0.01%
+ Complexity 33825 33474 -351
============================================
Files 1913 1913
Lines 149281 149312 +31
Branches 17325 17330 +5
============================================
+ Hits 110902 110949 +47
+ Misses 29540 29517 -23
- Partials 8839 8846 +7
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24799/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24799/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.28% <21.27%> (-0.13%)` | :arrow_down: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24799/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `22.79% <30.85%> (-0.02%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24799/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.84% <85.10%> (+0.03%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24799?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...ker/resourcegroup/ResourceGroupPublishLimiter.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fresourcegroup%2FResourceGroupPublishLimiter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9yZXNvdXJjZWdyb3VwL1Jlc291cmNlR3JvdXBQdWJsaXNoTGltaXRlci5qYXZh) | `88.88% <100.00%> (+1.93%)` | :arrow_up: |
| [...rg/apache/pulsar/broker/service/AbstractTopic.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FAbstractTopic.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0Fic3RyYWN0VG9waWMuamF2YQ==) | `88.42% <100.00%> (+0.06%)` | :arrow_up: |
| [...rg/apache/pulsar/broker/service/BrokerService.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FBrokerService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0Jyb2tlclNlcnZpY2UuamF2YQ==) | `83.62% <100.00%> (+0.31%)` | :arrow_up: |
| [...ava/org/apache/pulsar/broker/service/Producer.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FProducer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1Byb2R1Y2VyLmphdmE=) | `81.00% <ø> (-0.19%)` | :arrow_down: |
| [.../pulsar/broker/service/PublishRateLimiterImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FPublishRateLimiterImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1B1Ymxpc2hSYXRlTGltaXRlckltcGwuamF2YQ==) | `88.88% <100.00%> (+0.48%)` | :arrow_up: |
| [...org/apache/pulsar/broker/service/TransportCnx.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FTransportCnx.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1RyYW5zcG9ydENueC5qYXZh) | `50.00% <ø> (ø)` | |
| [...va/org/apache/pulsar/broker/service/ServerCnx.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FServerCnx.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1NlcnZlckNueC5qYXZh) | `72.48% <75.00%> (-0.10%)` | :arrow_down: |
| [...ulsar/broker/service/ServerCnxThrottleTracker.java](https://app.codecov.io/gh/apache/pulsar/pull/24799?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FServerCnxThrottleTracker.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1NlcnZlckNueFRocm90dGxlVHJhY2tlci5qYXZh) | `81.69% <81.53%> (+0.83%)` | :arrow_up: |
... and [71 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24799/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #24799:
URL: https://github.com/apache/pulsar/pull/24799
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker] Part-2 of PIP-434: Use ServerCnxThrottleTracker, instead of modifying channel.readable directly [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24799:
URL: https://github.com/apache/pulsar/pull/24799#discussion_r2443885747
##########
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).
- * <p>
- * 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.
- * <p>
- * 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.
+ *
+ * <p>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.
+ *
+ * <h3>Throttling Mechanism</h3>
+ * <p>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.
+ *
+ * <h3>Supported Throttling Types</h3>
+ * <ul>
+ * <li><b>Connection-level:</b> Max pending publish requests, outbound buffer limits</li>
+ * <li><b>Thread-level:</b> IO thread memory limits for in-flight publishing</li>
+ * <li><b>Topic-level:</b> Topic publish rate limiting</li>
+ * <li><b>Resource Group-level:</b> Resource group publish rate limiting</li>
+ * <li><b>Broker-level:</b> Global broker publish rate limiting</li>
+ * <li><b>Flow Control:</b> Channel writability and cooldown rate limiting</li>
+ * </ul>
+ *
+ * <h3>Reentrant vs Non-Reentrant Types</h3>
+ * <p>Some throttling types support multiple concurrent activations (reentrant):
+ * <ul>
+ * <li>{@link ThrottleType#TopicPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same topic</li>
+ * <li>{@link ThrottleType#ResourceGroupPublishRate} - Reentrant because multiple producers may share
+ * the same rate limiter which relates to the same resource group</li>
+ * </ul>
+ * <p>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.
+ *
+ * <h3>Thread Safety</h3>
+ * <p>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.
+ *
+ * <h3>Usage Example</h3>
+ * <pre>{@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);
+ * }</pre>
+ *
+ * @see ThrottleType
+ * @see ThrottleRes
+ * @see ServerCnx
*/
@Slf4j
-final class ServerCnxThrottleTracker {
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> THROTTLE_COUNT_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "throttleCount");
-
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker>
- PENDING_SEND_REQUESTS_EXCEEDED_UPDATER =
- AtomicIntegerFieldUpdater.newUpdater(
- ServerCnxThrottleTracker.class, "pendingSendRequestsExceeded");
- private static final AtomicIntegerFieldUpdater<ServerCnxThrottleTracker> 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.
+ *
+ * <p>Each type represents a specific resource constraint or rate limiting condition
+ * that may require throttling the connection to maintain system stability and fairness.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxPendingPublishRequestsPerConnection()}
+ */
+ ConnectionMaxQuantityOfInFlightPublishing,
+ /**
+ * Throttling due to excessive memory usage by in-flight publish operations on the IO thread.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getMaxMessagePublishBufferSizeInMB()}
+ */
+ IOThreadMaxBytesOfInFlightPublishing,
- public ServerCnxThrottleTracker(ServerCnx serverCnx) {
- this.serverCnx = serverCnx;
+ /**
+ * Throttling due to topic-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Topic-level publish rate policies
+ */
+ TopicPublishRate,
+
+ /**
+ * Throttling due to resource group-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Reentrant (supports multiple concurrent activations)
+ * <br><b>Reason for reentrancy:</b> 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.
+ *
+ * <p><b>Configuration:</b> Resource group publish rate policies
+ */
+ ResourceGroupPublishRate,
+ /**
+ * Throttling due to broker-level publish rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Configuration:</b> {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxMessageRate()}
+ * and {@link ServiceConfiguration#getBrokerPublisherThrottlingMaxByteRate()}
+ */
+ BrokerPublishRate,
+
+ /**
+ * Throttling due to channel outbound buffer being full.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ * <p><b>Reference:</b> PIP-434: Expose Netty channel configuration WRITE_BUFFER_WATER_MARK
+ */
+ ConnectionOutboundBufferFull,
+
+ /**
+ * Throttling due to connection pause/resume cooldown rate limiting.
+ *
+ * <p>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.
+ *
+ * <p><b>Type:</b> Non-reentrant
+ */
+ ConnectionPauseReceivingCooldownRateLimit
+ }
+
+ /**
+ * Enumeration representing the result of a throttling state change operation.
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>The connection transitions from unthrottled to throttled (first throttle type activated)</li>
+ * <li>The connection transitions from throttled to unthrottled (last throttle type deactivated)</li>
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>A throttle type is activated, but the connection was already throttled by other types</li>
+ * <li>A throttle type is deactivated, but the connection remains throttled by other types</li>
+ * <li>A reentrant throttle type's counter is incremented or decremented</li>
+ * </ul>
+ */
+ TypeStateChanged,
+
+ /**
+ * The operation was dropped because it would violate the throttle type's constraints.
+ *
+ * <p>This occurs when:
+ * <ul>
+ * <li>Attempting to mark a non-reentrant throttle type that is already active</li>
+ * <li>Attempting to unmark a throttle type that is not currently active</li>
+ * <li>Attempting to unmark a reentrant throttle type with an invalid counter state</li>
+ * </ul>
+ */
+ Dropped
}
/**
- * See {@link Producer#incrementThrottleCount()} for documentation.
+ * Checks if the connection is currently throttled by any throttle type.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
Review Comment:
This assertion breaks the protocol handler feature in https://github.com/streamnative/kop/pull/1589
The root cause is that the public `incrementPublishCount` method is called directly by the protocol handler but it's not guaranteed to be called in the `ServerCnx`'s event loop thread. In KoP, the `ServerCnx` object is just a mock object that does not serve any connection.
Currently, I made a workaround like:
```java
// TODO: Here we cannot leverage the `PersistentTopic#incrementPublishCount` method because it could call
// `markThrottled` and `unmarkThrottled` from `ServerCnxThrottleTracker`, which has a strong assertion on
// the thread that calls this method. See https://github.com/apache/pulsar/pull/24799 for details.
try {
final var topic = (AbstractTopic) producer.getTopic();
final var bytesInCounter = AbstractTopic.class.getDeclaredField("bytesInCounter");
bytesInCounter.setAccessible(true);
((LongAdder) bytesInCounter.get(topic)).add(readableBytes);
final var msgInCounter = AbstractTopic.class.getDeclaredField("msgInCounter");
msgInCounter.setAccessible(true);
((LongAdder) msgInCounter.get(topic)).add(numMessages);
} catch (NoSuchFieldException | ClassCastException | IllegalAccessException e) {
if (errorLoggedForReflection.compareAndSet(false, true)) {
log.error("Unable to update bytes and msg for topic", e);
}
}
```
Though the publish rate limiter will not be supported by the downstream protocol handler. /cc @tjiuming @Demogorgon314
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org