[fix][fn] Record Pulsar Function processing time properly for asynchronous functions #23811

walkinggo · 2025-01-04T09:51:57Z

Motivation

Modifications

@walkinggo Please update this. (You can use GenAI in assisting to summarize the modifications. Some hints in #23806 (comment) . GitHub's Co-pilot seems to support summarizing PR modifications too.)

Documentation

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

Matching PR in forked repository

PR in forked repository:

lhotari · 2025-01-04T12:30:31Z

pulsar-client/src/test/java/org/apache/pulsar/client/examples/TransactionExample.java

This change doesn't seem to be related to the PR.

Sorry, uploading these file changes was a mistake. I've force-pushed in a later commit to remove those changes.

lhotari · 2025-01-04T13:19:40Z

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

+                CompletableFuture<InstanceObserver> future = CompletableFuture.supplyAsync(() -> {
+                    JavaExecutionResult result;
+                    InstanceObserver instanceObserver = new InstanceObserver();
+                    instanceObserver.setStartTime(System.nanoTime());
+                    // process the message
+                    Thread.currentThread().setContextClassLoader(functionClassLoader);
+                    result = javaInstance.handleMessage(
+                            currentRecord,
+                            currentRecord.getValue(),
+                            asyncResultConsumer,
+                            asyncErrorHandler);
+                    Thread.currentThread().setContextClassLoader(instanceClassLoader);
+                    instanceObserver.setJavaExecutionResult(result);
+                    return instanceObserver;
+                }).whenComplete((res, ex) -> {
+                    stats.processTimeEnd(res.getStartTime());
+                });

+                JavaExecutionResult result = future.join().getJavaExecutionResult();


This doesn't make sense and is completely unnecessary.
Instead, you should simply pass the starting time nanos as an argument that gets added to the JavaExecutionResult instance field (add a new field startTimeNanos to JavaExecutionResult).
The result can be recorded in the handleResult method. The processTimeStart method should be completely removed from the ComponentStatsManager class and the processTimeEnd method should accept the starting time as a parameter. The value gets taken from the startTimeNanos field of the JavaExecutionResult instance.

Thank you, I have completed these tasks.

…m ComponentStatsManager to JavaExecutionResult.

walkinggo · 2025-01-04T17:18:13Z

It looks like due to code formatting, many unintended changes were introduced. I added startTimeNanos in JavaExecutionResult and removed it from ComponentStatsManager. Now, the processTimeEnd method in ComponentStatsManager takes the start time as a parameter and calculates the final time. I also removed the processTimeStart method.

walkinggo · 2025-01-04T17:19:51Z

Do I need to create a new PR to remove the unnecessary line changes caused by formatting? @ @lhotari

lhotari

Good start, going in the right direction!

lhotari · 2025-01-04T19:28:52Z

...instance/src/main/java/org/apache/pulsar/functions/instance/stats/ComponentStatsManager.java

-    public abstract void processTimeStart();
-
-    public abstract void processTimeEnd();
+    public abstract void processTimeEnd(double processTimeStart);


the parameter should be a long instead of double.

lhotari · 2025-01-04T19:30:53Z

...nctions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaExecutionResult.java

@@ -29,6 +29,7 @@
 public class JavaExecutionResult {
    private Throwable userException;
    private Object result;
+    private double startTimeNanos;


Suggested change

private double startTimeNanos;

private final long startTimeNanos = System.nanoTime();

It should be a long field and it should be made final.

lhotari · 2025-01-04T19:33:21Z

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

@@ -117,6 +117,7 @@ public JavaExecutionResult handleMessage(Record<?> record, Object input,
        }

        JavaExecutionResult executionResult = new JavaExecutionResult();
+        executionResult.setStartTimeNanos(System.nanoTime());


This is obsolete since startTimeNanos is set when the instance gets constructed.

You will also need to pass the JavaExecutionResult instance into a field javaExecutionResult of the AsyncFuncRequest so that the same instance is being used for the asyncPreserveInputOrderForOutputMessages case on line 139. Replace the instantiation of JavaExecutionResult in processAsyncResultsInInputOrder method on line 191 with this instance that is carried in the javaExecutionResult of the AsyncFuncRequest instance.
Also replace the instance on line 152 with this same executionResult on line 152 so that the starting time will be handled for the non asyncPreserveInputOrderForOutputMessages case of async processing.
With these changes, the solution would have the basics in place. There would also need to be a way to test all of this.

For testing, it might be useful to have a java.util.function.LongSupplier function for getting the nanoTime. In production code, this would default to () -> System.nanoTime(), but in tests, this could be used to mock the "clock" for getting nanoTime. Adding testability would cause some additional refactorings.

lhotari · 2025-01-04T19:34:24Z

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

+                    // register end time
+                    stats.processTimeEnd(result.getStartTimeNanos());


Remove this. This should be handled in the handleResult method instead.

lhotari · 2025-01-04T19:51:38Z

Do I need to create a new PR to remove the unnecessary line changes caused by formatting? @ @lhotari

It makes it slightly harder to read the diff when there are formatting changes. However there's the "hide whitespace" option to address that in GitHub UI.
Are you using IntelliJ/IDEA for development? Have you setup the code style according to the instructions at https://pulsar.apache.org/contribute/setup-ide/#configure-code-style ? I noticed that the import formatting doesn't match the Pulsar formatting since there are no empty lines in the Pulsar code style.
In the past, I have noticed that some code in the Pulsar Functions packages is using some non-default formatting so if you reformat the code, it will result in many changes. As long as you are using the Pulsar code style, the formatting changes are fine to be included in this PR.

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jan 4, 2025

walkinggo mentioned this pull request Jan 4, 2025

[Bug][pulsar-function]Pulsar Function processing time doesn't get properly recorded for asynchronous functions #23787

Closed

4 tasks

lhotari reviewed Jan 4, 2025

View reviewed changes

lhotari requested changes Jan 4, 2025

View reviewed changes

change the class JavaExecutionResult and move the startTime field fro…

3511c82

…m ComponentStatsManager to JavaExecutionResult.

walkinggo force-pushed the complete-get-proper-time branch from 58d57af to 3511c82 Compare January 4, 2025 17:10

lhotari requested changes Jan 4, 2025

View reviewed changes

lhotari changed the title ~~[Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture .~~ [fix][fn] Record Pulsar Function processing time properly for asynchronous functions Jan 4, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][fn] Record Pulsar Function processing time properly for asynchronous functions #23811

[fix][fn] Record Pulsar Function processing time properly for asynchronous functions #23811

walkinggo commented Jan 4, 2025 •

edited by lhotari

Loading

lhotari Jan 4, 2025

walkinggo Jan 4, 2025

lhotari Jan 4, 2025

walkinggo Jan 4, 2025

walkinggo commented Jan 4, 2025

walkinggo commented Jan 4, 2025

lhotari left a comment

lhotari Jan 4, 2025

lhotari Jan 4, 2025

lhotari Jan 4, 2025

lhotari Jan 4, 2025

lhotari Jan 4, 2025

lhotari commented Jan 4, 2025

	private double startTimeNanos;
	private final long startTimeNanos = System.nanoTime();

		// register end time
		stats.processTimeEnd(result.getStartTimeNanos());

[fix][fn] Record Pulsar Function processing time properly for asynchronous functions #23811

Are you sure you want to change the base?

[fix][fn] Record Pulsar Function processing time properly for asynchronous functions #23811

Conversation

walkinggo commented Jan 4, 2025 • edited by lhotari Loading

Motivation

Modifications

Documentation

Matching PR in forked repository

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

walkinggo commented Jan 4, 2025

walkinggo commented Jan 4, 2025

lhotari left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lhotari commented Jan 4, 2025

walkinggo commented Jan 4, 2025 •

edited by lhotari

Loading