You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "walkinggo (via GitHub)" <gi...@apache.org> on 2025/01/04 09:51:59 UTC
[PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
walkinggo opened a new pull request, #23811:
URL: https://github.com/apache/pulsar/pull/23811
<!--
### Contribution Checklist
- PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://pulsar.apache.org/contribute/develop-semantic-title/)*.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
<!-- Either this PR fixes an issue, -->
<!-- or this PR is one task of an issue -->
Main Issue: #23705
related pr: #23787
<!-- If the PR belongs to a PIP, please add the PIP link here -->
<!-- Details of when a PIP is required and how the PIP process work, please see: https://github.com/apache/pulsar/blob/master/pip/README.md -->
### Motivation
Use `CompletableFuture` to ensure that the function call time is correctly captured. Separate the time tracking variable from `ComponentStatsManager` entirely.
<!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. -->
### Modifications
Place the specific message processing logic inside the `CompletableFuture` to ensure the time is correctly captured.
<!-- Describe the modifications you've done. -->
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2571360424
Do I need to create a new PR to remove the unnecessary line changes caused by formatting? @ @lhotari
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906727569
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
It seems pointless to test that at all in this unit test class. It would essentially test that `result.getStartTime()` was set before the execution started and that execution took more than 500ms. Tests should be about testing expected behavior.
In this PR, we are more interested that the processing time gets properly recorded. That would have to be tested at a different level since JavaInstanceTest is at the level of the JavaInstance class. The processing time handling is in JavaInstanceRunnable. I think that tests for this PR should reside in JavaInstanceRunnable and possibly also in one of the end-to-end tests (for example `pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java`).
btw. One tip about assertions. For newly added tests, I'd recommend using [AssertJ fluent assertions](https://assertj.github.io/doc/) for more complex assertions. We have AssertJ already available as a test dependency in Pulsar. It's not that I'm saying that AssertJ must be used, it's just that if there are more complex assertions, it's a better option than TestNG's assertions. AssertJ assertions are easier to read and when there are failures, the error messages are really great.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1903092224
##########
pulsar-client/src/test/java/org/apache/pulsar/client/examples/TransactionExample.java:
##########
Review Comment:
This change doesn't seem to be related to the PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1913347928
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java:
##########
@@ -173,6 +178,20 @@ public Void process(String input, Context context) throws Exception {
}
}
+ @Test
+ public void testFunctionAsyncTime() throws Exception {
+ JavaInstanceRunnable runnable =
+ createRunnable(FunctionDetails.newBuilder().setClassName(TestFunction.class.getName()).build());
+ Instant before = Instant.now();
+ Instant join = CompletableFuture.supplyAsync(() -> {
+ new Thread(runnable).start();
+ return Instant.now();
+ }).join();
+ assertThat(join).isNotNull();
+ Duration duration = Duration.between(before, join);
+ assertThat(duration.toMillis()).isLessThan(20);
Review Comment:
I have made the changes. Is it okay now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1907195505
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
> How should I run `PulsarFunctionE2ETest.java` located in `pulsar-broker/src/test/java/org/apache/pulsar/io`? I tried running it locally in IDEA, but encountered an error. Is there any related documentation?
Most likely you will first have to compile Pulsar on the command line with maven.
either
build core-modules:
```shell
mvn -Pcore-modules,-main -T 1C clean install -DskipTests -Dspotbugs.skip=true
```
or build everything, skipping some checks to speed it up:
```shell
mvn -T 1C clean install -DskipTests -Dspotbugs.skip=true -Dcheckstyle.skip=true -Dlicense.skip=true
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1907361282
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
> Is this test program only runnable on Linux or macOS? I ran it on Windows using the command you provided, but encountered an error indicating that I don't have permission to delete certain files and that the path characters are incorrect.
@walkinggo I'd recommend using [Windows Subsystem for Linux (WSL) on Windows](https://learn.microsoft.com/en-us/windows/wsl/install) with Ubuntu for developing Pulsar. We don't support Windows for Pulsar broker at all. It might work, but the support is only for Linux and macOS. For the Pulsar client, we also support Windows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1907201227
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
Review Comment:
yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906749356
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
Review Comment:
Alright, I will remove this test case. Should I add additional test cases in the other test files you mentioned?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1907184157
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
How should I run `PulsarFunctionE2ETest.java` located in `pulsar-broker/src/test/java/org/apache/pulsar/io`? I tried running it locally in IDEA, but encountered an error. Is there any related documentation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906651064
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -24,12 +24,16 @@
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
+import java.util.function.LongSupplier;
Review Comment:
These seem to be unused imports
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
Review Comment:
This doesn't seem to be testing an async function since it's a `Function<String, String>` that is the function.
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
What is the logic behind this assertion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2580207178
I pass the start time of the result as a parameter rather than making it an inherent attribute of the result. This is because I don't want the result to become an inherent variable of `AsyncFuncRequest`. If it were an inherent variable, the result would be exposed, triggering the EI_EXPOSE_REP warning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece merged PR #23811:
URL: https://github.com/apache/pulsar/pull/23811
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1903143495
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:
##########
@@ -346,10 +348,10 @@ public void run() {
asyncErrorHandler);
Thread.currentThread().setContextClassLoader(instanceClassLoader);
- // register end time
- stats.processTimeEnd();
if (result != null) {
+ // register end time
+ stats.processTimeEnd(result.getStartTimeNanos());
Review Comment:
Remove this. This should be handled in the handleResult method instead.
##########
pulsar-functions/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;
Review Comment:
```suggestion
private final long startTimeNanos = System.nanoTime();
```
It should be a `long` field and it should be made final.
##########
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());
Review Comment:
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.
##########
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());
Review Comment:
This is obsolete since startTimeNanos is set when the instance gets constructed.
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/ComponentStatsManager.java:
##########
@@ -100,9 +102,7 @@ public ComponentStatsManager(FunctionCollectorRegistry collectorRegistry,
public abstract void setLastInvocation(long ts);
- public abstract void processTimeStart();
-
- public abstract void processTimeEnd();
+ public abstract void processTimeEnd(double processTimeStart);
Review Comment:
the parameter should be a `long` instead of `double`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906247641
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +314,22 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ Thread.sleep(500);
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ LongSupplier timeSupplier = () -> System.nanoTime() - 500_000_000L;
+ assertNotNull(result.getResult());
+ long beforeTime = timeSupplier.getAsLong();
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
Thank you for your patience. I have removed the `LongSupplier timeSupplier`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2579942805
Should I modify the related code logic based on the following error that appeared in my local CI check? @lhotari
[ERROR] Medium: org.apache.pulsar.functions.instance.JavaInstance$AsyncFuncRequest.getResult() may expose internal representation by returning JavaInstance$AsyncFuncRequest.result [org.apache.pulsar.functions.instance.JavaInstance$AsyncFuncRequest] At JavaInstance.java:[line 50] EI_EXPOSE_REP
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1903098213
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:
##########
@@ -329,26 +334,29 @@ public void run() {
}
}
- JavaExecutionResult result;
// set last invocation time
stats.setLastInvocation(System.currentTimeMillis());
- // 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();
+ 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();
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2571360085
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1907363754
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
IntelliJ has also special support for WSL on Windows: https://www.jetbrains.com/help/idea/how-to-use-wsl-development-environment-in-product.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2575589628
@lhotari Alright, I've used another branch to solve the issue. I'll now overwrite the current branch with the other one and close the next PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1907287557
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
Is this test program only runnable on Linux or macOS? I ran it on Windows using the command you provided, but encountered an error indicating that I don't have permission to delete certain files and that the path characters are incorrect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1930928278
##########
pulsar-functions/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 final long startTime = System.nanoTime();
public void reset() {
Review Comment:
I think it would be useful to remove this unused `reset` method from this class. The new final `startTime` field makes this method inconsistent as well and the resolution is to simply remove this method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2575534534
> I didn't set the correct code formatting in IDEA. Therefore, I will close this PR and open a new one to fix the issue.see #23820
@walkinggo There shouldn't be a need to create new PRs in this type of cases. You could have simply reformatted the code. All commits in the PR will get squashed into a single commit when the PR gets merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1905794777
##########
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());
Review Comment:
I have completed these tasks, please check them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906667087
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
Review Comment:
Sorry, I should have used the `CompletableFuture` class for testing. Is that acceptable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1905803095
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +314,22 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ Thread.sleep(500);
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ LongSupplier timeSupplier = () -> System.nanoTime() - 500_000_000L;
+ assertNotNull(result.getResult());
+ long beforeTime = timeSupplier.getAsLong();
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
the usage of `LongSupplier timeSupplier` doesn't make much sense in this way. just get rid of it for now. What I was trying to explain earlier was that a common pattern for testing time is to have a mock clock that gets injected also in the production code. Instead of calling `System.nanoTime()` in production code, the "clock" interface (`LongSupplier` for example) would be used. Let's forget that for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2581564524
I have added the content to the corresponding file. Thank you for your help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1931454828
##########
pulsar-functions/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 final long startTime = System.nanoTime();
public void reset() {
Review Comment:
Thank you for your help and patience. I really appreciate the time and effort you've spent on my PR. I have removed the method. Could you please review it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1923840366
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
Review Comment:
Please don't format this file to avoid conflict when this PR is cherry-picked to the other version.
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java:
##########
@@ -202,7 +234,8 @@ public void testFunctionResultNull() throws Exception {
@NotNull
private JavaInstanceRunnable getJavaInstanceRunnable(boolean autoAck,
- org.apache.pulsar.functions.proto.Function.ProcessingGuarantees processingGuarantees) throws Exception {
+ org.apache.pulsar.functions.proto.Function.ProcessingGuarantees processingGuarantees)
Review Comment:
Please revert these changes.
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:
##########
Review Comment:
DO NOT format the unmodified code.
##########
pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java:
##########
Review Comment:
DO NOT format the unmodified code.
##########
pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java:
##########
@@ -1204,4 +1307,12 @@ public ByteBuffer process(ByteBuffer input, Context context) throws Exception {
return input;
}
}
+
+ public static class SpendTimeFunction implements org.apache.pulsar.functions.api.Function<String, String> {
Review Comment:
`org.apache.pulsar.functions.api.examples.JavaNativeAsyncExclamationFunction` can be instead of this function, please remove this class.
BTW, this is a sync function, not async.
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java:
##########
@@ -173,6 +184,27 @@ public Void process(String input, Context context) throws Exception {
}
}
+ @Test
+ public void testFunctionAsyncTime() throws Exception {
+ FunctionDetails functionDetails = FunctionDetails.newBuilder()
+ .setAutoAck(true)
+ .setProcessingGuarantees(org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.MANUAL)
+ .build();
+ JavaInstanceRunnable javaInstanceRunnable = createRunnable(functionDetails);
+ Field stats = JavaInstanceRunnable.class.getDeclaredField("stats");
Review Comment:
DO NOT use the Java reflection to set the stats.
Please add a set method to the `JavaInstanceRunnable`:
```
@VisibleForTesting
void setStats(ComponentStatsManager stats) {
this.stats = stats;
}
```
And then call this method to set the `manager` as the `stats` value.
##########
pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java:
##########
@@ -296,6 +305,92 @@ public void testReadCompactedFunction() throws Exception {
producer.close();
}
+ @Test(timeOut = 20000)
+ public void testPulsarFunctionAsyncStatTime() throws Exception {
+ final String namespacePortion = "io";
+ final String replNamespace = tenant + "/" + namespacePortion;
+ final String sourceTopic = "persistent://" + replNamespace + "/my-topic1";
+ final String sinkTopic = "persistent://" + replNamespace + "/output";
+ final String propertyKey = "key";
+ final String propertyValue = "value";
+ final String functionName = SpendTimeFunction.class.getName();
Review Comment:
```suggestion
final String functionName = "org.apache.pulsar.functions.api.examples.JavaNativeAsyncExclamationFunction";
```
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/ComponentStatsManager.java:
##########
Review Comment:
DO NOT format the unmodified code.
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java:
##########
Review Comment:
DO NOT format the unmodified code.
##########
pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java:
##########
@@ -296,6 +305,92 @@ public void testReadCompactedFunction() throws Exception {
producer.close();
}
+ @Test(timeOut = 20000)
+ public void testPulsarFunctionAsyncStatTime() throws Exception {
+ final String namespacePortion = "io";
+ final String replNamespace = tenant + "/" + namespacePortion;
+ final String sourceTopic = "persistent://" + replNamespace + "/my-topic1";
+ final String sinkTopic = "persistent://" + replNamespace + "/output";
+ final String propertyKey = "key";
+ final String propertyValue = "value";
Review Comment:
```suggestion
```
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java:
##########
@@ -173,6 +184,27 @@ public Void process(String input, Context context) throws Exception {
}
}
+ @Test
+ public void testFunctionAsyncTime() throws Exception {
+ FunctionDetails functionDetails = FunctionDetails.newBuilder()
+ .setAutoAck(true)
+ .setProcessingGuarantees(org.apache.pulsar.functions.proto.Function.ProcessingGuarantees.MANUAL)
+ .build();
+ JavaInstanceRunnable javaInstanceRunnable = createRunnable(functionDetails);
+ Field stats = JavaInstanceRunnable.class.getDeclaredField("stats");
+ FunctionStatsManager manager = mock(FunctionStatsManager.class);
+ stats.setAccessible(true);
+ stats.set(javaInstanceRunnable, manager);
+ stats.setAccessible(false);
+ JavaExecutionResult javaExecutionResult = new JavaExecutionResult();
+ Thread.sleep(500);
+ Record record = mock(Record.class);
+ javaInstanceRunnable.handleResult(record, javaExecutionResult);
+ ArgumentCaptor<Long> timeCaptor = ArgumentCaptor.forClass(Long.class);
+ verify(manager).processTimeEnd(timeCaptor.capture());
+ Assert.assertEquals(timeCaptor.getValue(), javaExecutionResult.getStartTime() + 500, 100);
Review Comment:
```suggestion
Assert.assertEquals(timeCaptor.getValue(), javaExecutionResult.getStartTime());
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java:
##########
@@ -93,6 +95,8 @@ protected static FunctionConfig createFunctionConfig(String tenant, String names
functionConfig.setAutoAck(true);
if (!isBuiltin) {
functionConfig.setClassName("org.apache.pulsar.functions.api.examples.ExclamationFunction");
+ } else {
+ functionConfig.setClassName(functionName);
Review Comment:
Why add this line?
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java:
##########
@@ -255,10 +288,10 @@ public void testSourceConfigParsingPreservesOriginalType() throws Exception {
public Object[][] component() {
return new Object[][]{
// Schema: component type, whether to map in secrets
- { FunctionDetails.ComponentType.SINK },
- { FunctionDetails.ComponentType.SOURCE },
- { FunctionDetails.ComponentType.FUNCTION },
- { FunctionDetails.ComponentType.UNKNOWN },
+ {FunctionDetails.ComponentType.SINK},
Review Comment:
Please revert these changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1903130446
##########
pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java:
##########
@@ -329,26 +334,29 @@ public void run() {
}
}
- JavaExecutionResult result;
// set last invocation time
stats.setLastInvocation(System.currentTimeMillis());
- // 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();
+ 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();
Review Comment:
Thank you, I have completed these tasks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2571395788
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo closed pull request #23811: [fix][fn] Record Pulsar Function processing time properly for asynchronous functions
URL: https://github.com/apache/pulsar/pull/23811
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906663116
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -24,12 +24,16 @@
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
+import java.util.function.LongSupplier;
Review Comment:
Sorry, the import statements here were left over from my coding, and I should have removed them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1913037568
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java:
##########
@@ -173,6 +178,20 @@ public Void process(String input, Context context) throws Exception {
}
}
+ @Test
+ public void testFunctionAsyncTime() throws Exception {
+ JavaInstanceRunnable runnable =
+ createRunnable(FunctionDetails.newBuilder().setClassName(TestFunction.class.getName()).build());
+ Instant before = Instant.now();
+ Instant join = CompletableFuture.supplyAsync(() -> {
+ new Thread(runnable).start();
+ return Instant.now();
+ }).join();
+ assertThat(join).isNotNull();
+ Duration duration = Duration.between(before, join);
+ assertThat(duration.toMillis()).isLessThan(20);
Review Comment:
this assertion seems pointless. It would be more meaningful to validate calls to `processTimeEnd`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2581217946
> I pass the start time of the result as a parameter rather than making it an inherent attribute of the result. This is because I don't want the result to become an inherent variable of `AsyncFuncRequest`. If it were an inherent variable, the result would be exposed, triggering the EI_EXPOSE_REP warning.
That makes the code worse in my opinion. I'd recommend adding a definition to the `src/main/resources/findbugsExclude.xml` file to ignore the `EI_EXPOSE_REP` SpotBugs/FindBugs check for the particular method. We do that already for many other locations. This is the file where you can add a new entry: https://github.com/apache/pulsar/blob/master/pulsar-functions/instance/src/main/resources/findbugsExclude.xml .
```xml
<Match>
<Class name="org.apache.pulsar.functions.instance.JavaInstance$AsyncFuncRequest"/>
<Method name="getResult"/>
<Bug pattern="EI_EXPOSE_REP"/>
</Match>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Bug][pulsar-function]Solve pulsar Function processing time doesn't get properly recorded for asynchronous functions with completefuture . [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1903130348
##########
pulsar-client/src/test/java/org/apache/pulsar/client/examples/TransactionExample.java:
##########
Review Comment:
Sorry, uploading these file changes was a mistake. I've force-pushed in a later commit to remove those changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1924819845
##########
pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java:
##########
@@ -93,6 +95,8 @@ protected static FunctionConfig createFunctionConfig(String tenant, String names
functionConfig.setAutoAck(true);
if (!isBuiltin) {
functionConfig.setClassName("org.apache.pulsar.functions.api.examples.ExclamationFunction");
+ } else {
+ functionConfig.setClassName(functionName);
Review Comment:
I added this line to modify the internal function name so that I could use the `SpendTimeFunction` method in the test cases. However, I plan to remove this method and use `org.apache.pulsar.functions.api.examples.JavaNativeAsyncExclamationFunction` instead. Therefore, I will revert this modification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#issuecomment-2575528433
I didn't set the correct code formatting in IDEA. Therefore, I will close this PR and open a new one to fix the issue.see #23820
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906665389
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
The logic here is to give a task that takes 500ms, and then compare the difference between the task's completion time and the recorded time. If the difference is less than 20ms, we can consider the recorded time as accurate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1906730580
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
Review Comment:
This test can be removed completely. More context in https://github.com/apache/pulsar/pull/23811#discussion_r1906727569 . Tests for this PR should go in different classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][fn] Record Pulsar Function processing time properly for asynchronous functions [pulsar]
Posted by "walkinggo (via GitHub)" <gi...@apache.org>.
walkinggo commented on code in PR #23811:
URL: https://github.com/apache/pulsar/pull/23811#discussion_r1907433793
##########
pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java:
##########
@@ -309,4 +317,25 @@ public void testAsyncFunctionMaxPendingVoidResult() throws Exception {
log.info("start:{} end:{} during:{}", startTime, endTime, endTime - startTime);
instance.close();
}
+
+ @Test
+ public void testAsyncFunctionTime() {
+ JavaInstance instance = new JavaInstance(
+ mock(ContextImpl.class),
+ (Function<String, String>) (input, context) -> {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return input;
+ },
+ new InstanceConfig());
+ String testString = "ABC123";
+ JavaExecutionResult result = instance.handleMessage(mock(Record.class), testString);
+ assertNotNull(result.getResult());
+ long beforeTime = System.nanoTime() - 500_000_000L;
+ assertTrue(Math.abs(beforeTime - result.getStartTime()) <= 20_000_000);
Review Comment:
Thank you for your patience. I have successfully run the program on Ubuntu. @lhotari
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org