Summary

Adds OpenTelemetry metrics to expose the status of the producer memory buffer defined by producerMemoryLimit in the Java client. This provides visibility into memory usage and helps with monitoring and alerting on client-side memory consumption.

Motivation

Currently, there's no way to monitor the memory buffer usage of Pulsar Java clients. Users cannot:

• Track how much memory is being consumed by pending messages
• Set up alerts when memory usage approaches the configured limit
• Debug memory-related issues or performance bottlenecks
• Monitor memory efficiency across different client configurations

Changes

New Metrics

Two new OpenTelemetry metrics are added:

1. pulsar.client.memory.buffer.usage (bytes)

   • Current memory buffer usage by the client
   • Updates in real-time as messages are queued and sent

2. pulsar.client.memory.buffer.limit (bytes)

   • Memory buffer limit configured for the client
   • Constant value based on client configuration

Key Features

• Conditional creation: Metrics are only created when memory limiting is enabled (memoryLimitBytes > 0)
• Zero overhead: When memory limiting is disabled, no metrics are registered
• Observable metrics: Automatically report current values via callbacks
• Existing patterns: Follows established OpenTelemetry infrastructure in the codebase

Implementation Details

• New ObservableUpDownCounter class that extends existing metrics infrastructure
• New MemoryBufferStats class manages the observable metrics
• Extended InstrumentProvider with observable counter support
• Added memoryLimit() method to MemoryLimitController for metric access
• Conditional metrics creation in PulsarClientImpl

Usage Example

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .memoryLimit(64, SizeUnit.MEGA_BYTES)
    .openTelemetry(openTelemetry) // Your OpenTelemetry instance
    .build();

The metrics will be available as:

pulsar.client.memory.buffer.usage = 1048576  # Current usage in bytes
pulsar.client.memory.buffer.limit = 67108864 # 64MB limit in bytes

Testing

• Unit tests: MemoryBufferStatsTest with InMemoryMetricReader to verify metric values
• Integration tests: ClientMetricsTest with real broker to test end-to-end behavior
• Edge cases: Tests for both enabled and disabled memory limiting scenarios

Verifying this change

• Unit tests pass
• Integration tests pass
• Checkstyle validation passes
• No performance regression when feature is disabled
• Backward compatibility maintained

Backward Compatibility

• Fully backward compatible - no breaking changes
• Metrics are only created when both OpenTelemetry and memory limiting are configured
• No performance impact when OpenTelemetry is not configured

Documentation

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