[Enhancement] PIP-234: Add a solution to share the memory limit controller solution

[lhotari]

Search before reporting

• I searched in the issues and found nothing similar.

Motivation

PIP-234 enables sharing of Pulsar client resources across multiple isolated Pulsar client instances. This was implemented in #24784 and #24790.
The solution lacks limiting the total amount of memory across all of the Pulsar client instances. That would be useful when memory limiting relies on the memory limit controller.

Solution

Modify org.apache.pulsar.client.api.PulsarClientSharedResources interface and implementation to support sharing memory limit controller in the same way as other shared resources are configured without directly referencing implementation instances.
Modify the current MemoryLimitController to support sharing. It would require support for registering multiple trigger callbacks and deregistering the callback when the client instance gets closed.

Alternatives

No response

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[oneby-wang]

I'm wondering how to deal with MemoryBufferStats in multiple Pulsar clients? Should they also be shared across multiple clients? I think MemoryBufferStats and MemoryLimitController should have a one-to-one relationship, but I'm not sure if my understanding is correct.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

Lines 289 to 297 in 6d415c6

	memoryLimitController = new MemoryLimitController(conf.getMemoryLimitBytes(),
	(long) (conf.getMemoryLimitBytes() * THRESHOLD_FOR_CONSUMER_RECEIVER_QUEUE_SIZE_SHRINKING),
	this::reduceConsumerReceiverQueueSize);
	// Only create memory buffer metrics if memory limiting is enabled
	if (memoryLimitController.isMemoryLimited()) {
	memoryBufferStats = new MemoryBufferStats(instrumentProvider, memoryLimitController);
	} else {
	memoryBufferStats = null;
	}

I'm not very familiar with OpenTelemetry, seems all the MemoryBufferStats instances use the same meter name in the master branch? Would that be a problem?

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/metrics/MemoryBufferStats.java

Lines 26 to 56 in 3dc1fd6

	public static final String BUFFER_USAGE_COUNTER = "pulsar.client.memory.buffer.usage";
	private final ObservableUpDownCounter bufferUsageCounter;
	public static final String BUFFER_LIMIT_COUNTER = "pulsar.client.memory.buffer.limit";
	private final ObservableUpDownCounter bufferLimitCounter;
	public MemoryBufferStats(InstrumentProvider instrumentProvider, MemoryLimitController memoryLimitController) {
	bufferUsageCounter = instrumentProvider.newObservableUpDownCounter(
	BUFFER_USAGE_COUNTER,
	Unit.Bytes,
	"Current memory buffer usage by the client",
	null, // no topic
	null, // no extra attributes
	measurement -> {
	if (memoryLimitController.isMemoryLimited()) {
	measurement.record(memoryLimitController.currentUsage());
	}
	});
	bufferLimitCounter = instrumentProvider.newObservableUpDownCounter(
	BUFFER_LIMIT_COUNTER,
	Unit.Bytes,
	"Memory buffer limit configured for the client",
	null, // no topic
	null, // no extra attributes
	measurement -> {
	if (memoryLimitController.isMemoryLimited()) {
	measurement.record(memoryLimitController.memoryLimit());
	}
	});
	}

[oneby-wang]

If this issue is not assigned, I'd like to contribute.

However, I'm about to go on vacation for about 20 days starting this Saturday. I'll start working on it once I'm back.

If the timeline is ok, I'd be very willing to work on this enhancement.

[lhotari]

I'm wondering how to deal with MemoryBufferStats in multiple Pulsar clients? Should they also be shared across multiple clients? I think MemoryBufferStats and MemoryLimitController should have a one-to-one relationship, but I'm not sure if my understanding is correct.

Yes, that's correct. org.apache.pulsar.client.impl.PulsarClientSharedResourcesImpl would also have to hold the MemoryBufferStats instance.

I'm not very familiar with OpenTelemetry, seems all the MemoryBufferStats instances use the same meter name in the master branch? Would that be a problem?

I don't think it's a problem, however it would require allowing to pass a org.apache.pulsar.client.impl.metrics.InstrumentProvider instance to the PulsarClientSharedResources builder so that the MemoryBufferStats could be configured appropriately. Right now the situation seems to be that there doesn't seem to be a way to distinguish values when there are multiple Pulsar client instances in a single JVM. I'd assume that there should be a way to configure common Otel attributes in InstrumentProvider, but that's missing. Since it's missing, I think that the concern could be ignored initially and just ensuring that the instrument provider given to PulsarClientSharedResources builder would be used. I think that the instrument provider instance should also be set to the pulsar client builder unless it's already been set.

[oneby-wang]

Oh, I see, it's the developer's duty to pass a different OpenTelemetry instance into PulsarClientSharedResourcesBuilder, just like the ClientBuilderImpl#openTelemetry() method.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java

Lines 154 to 158 in 3dc1fd6

	@Override
	public ClientBuilder openTelemetry(OpenTelemetry openTelemetry) {
	conf.setOpenTelemetry(openTelemetry);
	return this;
	}

Every ClientBuilderImpl and PulsarClientSharedResourcesBuilder should use different OpenTelemetry instances, otherwise they will shared the same global OpenTelemetry instance. Is this understanding correct?

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/metrics/InstrumentProvider.java

Lines 38 to 52 in 3dc1fd6

	public InstrumentProvider(OpenTelemetry otel) {
	if (otel == null) {
	// By default, metrics are disabled, unless the OTel java agent is configured.
	// This allows to enable metrics without any code change.
	otel = GlobalOpenTelemetry.get();
	}
	this.meter = otel.getMeterProvider()
	.meterBuilder("org.apache.pulsar.client")
	.setInstrumentationVersion(PulsarVersion.getVersion())
	.build();
	this.tracer = otel.getTracerProvider()
	.tracerBuilder("org.apache.pulsar.client")
	.setInstrumentationVersion(PulsarVersion.getVersion())
	.build();
	}

But, using the same global OpenTelemetry instance across multiple MemoryBufferStats instances seems to cause them to affect each other?

[lhotari]

Oh, I see, it's the developer's duty to pass a different OpenTelemetry instance into PulsarClientSharedResourcesBuilder, just like the ClientBuilderImpl#openTelemetry() method.

Yes. The original reason here was to be able to configure the underlying shared MemoryBufferStats instance.

Every ClientBuilderImpl and PulsarClientSharedResourcesBuilder should use different OpenTelemetry instances, otherwise they will shared the same global OpenTelemetry instance. Is this understanding correct?
But, using the same global OpenTelemetry instance across multiple MemoryBufferStats instances seems to cause them to affect each other?

Yes, it would be possible to configure the OpenTelemetry instances differently. However, that's very clumbersome and usually the global instance is shared. The limitation in the org.apache.pulsar.client.impl.metrics.InstrumentProvider class is that there's no way to scope the instance further so that different client instances could produce different type of metrics (let's say with different attributes).

I'm not sure how much of a problem it is to share an instance. Perhaps it's not. I'm sure that the source information would be missing so that you couldn't distinguish the source of a particular metric. It's also possible that there would be some clashes if the same instance is used to create a duplicate metric with the same attributes. However I think that solving that problem could be postponed since it's already possible to manually configure different OpenTelemetry/InstrumentProvider instances for the clients and have them configured in a way where the metrics don't clash.