PIP-442

Motivation

See PIP-442.
Apache Pulsar brokers currently lack memory limits for topic listing operations (CommandGetTopicsOfNamespace and CommandWatchTopicList), creating a significant gap in the broker's otherwise comprehensive memory management framework. This can lead to:

• Broker/Proxy Memory Exhaustion: Large namespaces with thousands of topics can trigger OutOfMemoryErrors when multiple clients simultaneously request topic lists
• Unpredictable Resource Usage: Operators cannot reliably predict or limit total memory consumption due to this unbounded allocation path
• Performance Degradation: Large topic list operations cause GC pressure and latency spikes affecting all broker operations

While Pulsar has robust memory limits for message publishing (maxMessagePublishBufferSizeInMB), managed ledger operations (managedLedgerMaxReadsInFlightSizeInMB, managedLedgerCacheSizeMB), and concurrent requests (maxConcurrentLookupRequest), topic listing operations remain unbounded.

Modifications

This PR implements PIP-442 by introducing memory-aware flow control for topic listing operations:

Core Components

1. AsyncSemaphore Interface & Implementation

   • Generic asynchronous semaphore abstraction with configurable timeouts and queue limits
   • Supports permit acquisition, updates, and cancellation
   • Tracks queue size and wait times for monitoring

2. AsyncDualMemoryLimiter Interface & Implementation

   • Dual memory pool tracking for heap memory (topic list assembly) and direct memory (network buffers)
   • Separate limits and queues for each memory type
   • Helper methods for common patterns (acquire-execute-release)

3. AsyncDualMemoryLimiterUtil

   • Utility methods for acquiring permits and writing responses
   • Handles serialization and network buffer management
   • Ensures permits are released after write completion or failure

4. TopicListMemoryLimiter

   • Extends AsyncDualMemoryLimiterImpl with metrics integration
   • Exposes both Prometheus and OpenTelemetry metrics
   • Tracks memory usage, queue sizes, wait times, and timeouts

Integration Points

Broker Integration:

• Modified ServerCnx.handleGetTopicsOfNamespace() to acquire initial heap permits (1KB estimate), update to actual size after topic retrieval, then acquire direct memory permits for response serialization
• Modified PulsarCommandSenderImpl to acquire direct memory permits before serialization and release after write completion
• Created TopicListMemoryLimiter in BrokerService with configured limits

Proxy Integration:

• Modified LookupProxyHandler.handleGetTopicsOfNamespace() with similar permit acquisition flow
• Handles both heap memory (for topic list from broker) and direct memory (for response to client)
• Created TopicListMemoryLimiter in ProxyService with configured limits

Configuration

Added six new configuration parameters to broker.conf, proxy.conf, and standalone.conf:

maxTopicListInFlightHeapMemSizeMB=100
maxTopicListInFlightDirectMemSizeMB=100
maxTopicListInFlightHeapMemSizePermitsAcquireTimeoutMillis=25000
maxTopicListInFlightHeapMemSizePermitsAcquireQueueSize=10000
maxTopicListInFlightDirectMemSizePermitsAcquireTimeoutMillis=25000
maxTopicListInFlightDirectMemSizePermitsAcquireQueueSize=10000

Metrics

Added 12 new metrics per instance (broker/proxy) with both Prometheus and OpenTelemetry support:

• Memory usage and limits (heap and direct)
• Queue sizes and limits
• Wait time summaries (P50, P95, P99, Max)
• Timeout counters

Verifying this change

• Make sure that the change passes the CI checks

This change added tests and can be verified as follows:

• Added PatternConsumerBackPressureMultipleConsumersTest: Integration test that creates 8,192 partitioned topics and sends 500 concurrent getTopicsOfNamespace requests from 200 different client connections. The test intentionally reduces available direct memory to reproduce memory pressure scenarios and verifies all requests complete successfully with memory limiting in place.

• Added ProxyPatternConsumerBackPressureMultipleConsumersTest: Extended version of the broker test that routes requests through the Pulsar Proxy to verify memory limiting works correctly in proxy scenarios.

• Added AsyncSemaphoreImplTest: Comprehensive unit tests for the async semaphore implementation covering:

  • Basic acquire/release operations
  • Queueing behavior when permits unavailable
  • Timeout and queue full exceptions
  • Cancellation support
  • Permit updates (increase/decrease)
  • Concurrent operations

• Added AsyncDualMemoryLimiterImplTest: Unit tests for the dual memory limiter covering:

  • Independent heap and direct memory limiting
  • Permit acquisition and updates for both types
  • Queue behavior and timeout handling
  • Concurrent operations with mixed memory types
  • Proper resource cleanup

• Added AsyncDualMemoryLimiterUtilTest: Tests for utility methods covering:

  • Successful permit acquisition and response writing
  • Error handling for permit acquisition failures
  • Write failures and proper permit release
  • Cancellation scenarios
  • Concurrent operations

• Verified metrics: Tests validate that all 26 metrics are properly registered and report correct values.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete