[Bug] Pulsar Function processing time doesn't get properly recorded for asynchronous functions

[lhotari]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

any released version

Minimal reproduce step

In the code, it can be seen that asynchronous functions don't get handled properly. stats.processTimeEnd() gets called immediately when the result is returned:

pulsar/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java

Lines 337 to 355 in 6fe8100

	// start time for process latency stat
	stats.processTimeStart();
	// process the message
	Thread.currentThread().setContextClassLoader(functionClassLoader);
	result = javaInstance.handleMessage(
	currentRecord,
	currentRecord.getValue(),
	asyncResultConsumer,
	asyncErrorHandler);
	Thread.currentThread().setContextClassLoader(instanceClassLoader);
	// register end time
	stats.processTimeEnd();
	if (result != null) {
	// process the synchronous results
	handleResult(currentRecord, result);
	}

What did you expect to see?

asynchronous functions would also be handled

What did you see instead?

asynchronous functions have invalid processing time stats

Anything else?

The current processing metric for async functions includes the time for doing the async calls and the waiting time when the concurrency limit is reached. The metric is useful for this purpose.
It doesn't tell the actual end-to-completion processing time which contains the async processing time.

Are you willing to submit a PR?

• I'm willing to submit a PR!